Untitled

mail@pastecode.io avatar
unknown
java
a year ago
35 kB
1
Indexable
Never
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//

package org.apache.kafka.clients.producer;

import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.kafka.clients.ApiVersions;
import org.apache.kafka.clients.ClientDnsLookup;
import org.apache.kafka.clients.ClientUtils;
import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.NetworkClient;
import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.internals.BufferPool;
import org.apache.kafka.clients.producer.internals.ProducerInterceptors;
import org.apache.kafka.clients.producer.internals.ProducerMetadata;
import org.apache.kafka.clients.producer.internals.ProducerMetrics;
import org.apache.kafka.clients.producer.internals.RecordAccumulator;
import org.apache.kafka.clients.producer.internals.Sender;
import org.apache.kafka.clients.producer.internals.TransactionManager;
import org.apache.kafka.clients.producer.internals.TransactionalRequestResult;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.errors.ApiException;
import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.common.errors.InvalidTopicException;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.errors.RecordTooLargeException;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.internals.ClusterResourceListeners;
import org.apache.kafka.common.metrics.JmxReporter;
import org.apache.kafka.common.metrics.KafkaMetricsContext;
import org.apache.kafka.common.metrics.MetricConfig;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.MetricsContext;
import org.apache.kafka.common.metrics.MetricsReporter;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.metrics.Sensor.RecordingLevel;
import org.apache.kafka.common.network.ChannelBuilder;
import org.apache.kafka.common.network.Selector;
import org.apache.kafka.common.record.AbstractRecords;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.common.utils.KafkaThread;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;

public class KafkaProducer<K, V> implements Producer<K, V> {
    private final Logger log;
    private static final String JMX_PREFIX = "kafka.producer";
    public static final String NETWORK_THREAD_PREFIX = "kafka-producer-network-thread";
    public static final String PRODUCER_METRIC_GROUP_NAME = "producer-metrics";
    private final String clientId;
    final Metrics metrics;
    private final Partitioner partitioner;
    private final int maxRequestSize;
    private final long totalMemorySize;
    private final ProducerMetadata metadata;
    private final RecordAccumulator accumulator;
    private final Sender sender;
    private final Thread ioThread;
    private final CompressionType compressionType;
    private final Sensor errors;
    private final Time time;
    private final Serializer<K> keySerializer;
    private final Serializer<V> valueSerializer;
    private final ProducerConfig producerConfig;
    private final long maxBlockTimeMs;
    private final ProducerInterceptors<K, V> interceptors;
    private final ApiVersions apiVersions;
    private final TransactionManager transactionManager;

    public KafkaProducer(Map<String, Object> configs) {
        this((Map)configs, (Serializer)null, (Serializer)null);
    }

    public KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
        this(new ProducerConfig(ProducerConfig.appendSerializerToConfig(configs, keySerializer, valueSerializer)), keySerializer, valueSerializer, (ProducerMetadata)null, (KafkaClient)null, (ProducerInterceptors)null, Time.SYSTEM);
    }

    public KafkaProducer(Properties properties) {
        this((Properties)properties, (Serializer)null, (Serializer)null);
    }

    public KafkaProducer(Properties properties, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
        this(Utils.propsToMap(properties), keySerializer, valueSerializer);
    }

    KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer, ProducerMetadata metadata, KafkaClient kafkaClient, ProducerInterceptors<K, V> interceptors, Time time) {
        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);
        }
    }

    Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadata metadata) {
        int maxInflightRequests = configureInflightRequests(this.producerConfig);
        int requestTimeoutMs = this.producerConfig.getInt("request.timeout.ms");
        ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(this.producerConfig, this.time, logContext);
        ProducerMetrics metricsRegistry = new ProducerMetrics(this.metrics);
        Sensor throttleTimeSensor = Sender.throttleTimeSensor(metricsRegistry.senderMetrics);
        KafkaClient client = kafkaClient != null ? kafkaClient : new NetworkClient(new Selector(this.producerConfig.getLong("connections.max.idle.ms"), this.metrics, this.time, "producer", channelBuilder, logContext), metadata, this.clientId, maxInflightRequests, this.producerConfig.getLong("reconnect.backoff.ms"), this.producerConfig.getLong("reconnect.backoff.max.ms"), this.producerConfig.getInt("send.buffer.bytes"), this.producerConfig.getInt("receive.buffer.bytes"), requestTimeoutMs, this.producerConfig.getLong("socket.connection.setup.timeout.ms"), this.producerConfig.getLong("socket.connection.setup.timeout.max.ms"), ClientDnsLookup.forConfig(this.producerConfig.getString("client.dns.lookup")), this.time, true, this.apiVersions, throttleTimeSensor, logContext);
        short acks = configureAcks(this.producerConfig, this.log);
        return new Sender(logContext, (KafkaClient)client, metadata, this.accumulator, maxInflightRequests == 1, this.producerConfig.getInt("max.request.size"), acks, this.producerConfig.getInt("retries"), metricsRegistry.senderMetrics, this.time, requestTimeoutMs, this.producerConfig.getLong("retry.backoff.ms"), this.transactionManager, this.apiVersions);
    }

    private static int lingerMs(ProducerConfig config) {
        return (int)Math.min(config.getLong("linger.ms"), 2147483647L);
    }

    private static int configureDeliveryTimeout(ProducerConfig config, Logger log) {
        int deliveryTimeoutMs = config.getInt("delivery.timeout.ms");
        int lingerMs = lingerMs(config);
        int requestTimeoutMs = config.getInt("request.timeout.ms");
        int lingerAndRequestTimeoutMs = (int)Math.min((long)lingerMs + (long)requestTimeoutMs, 2147483647L);
        if (deliveryTimeoutMs < lingerAndRequestTimeoutMs) {
            if (config.originals().containsKey("delivery.timeout.ms")) {
                throw new ConfigException("delivery.timeout.ms should be equal to or larger than linger.ms + request.timeout.ms");
            }

            deliveryTimeoutMs = lingerAndRequestTimeoutMs;
            log.warn("{} should be equal to or larger than {} + {}. Setting it to {}.", new Object[]{"delivery.timeout.ms", "linger.ms", "request.timeout.ms", lingerAndRequestTimeoutMs});
        }

        return deliveryTimeoutMs;
    }

    private TransactionManager configureTransactionState(ProducerConfig config, LogContext logContext) {
        TransactionManager transactionManager = null;
        boolean userConfiguredIdempotence = config.originals().containsKey("enable.idempotence");
        boolean userConfiguredTransactions = config.originals().containsKey("transactional.id");
        if (userConfiguredTransactions && !userConfiguredIdempotence) {
            this.log.info("Overriding the default {} to true since {} is specified.", "enable.idempotence", "transactional.id");
        }

        if (config.idempotenceEnabled()) {
            String transactionalId = config.getString("transactional.id");
            int transactionTimeoutMs = config.getInt("transaction.timeout.ms");
            long retryBackoffMs = config.getLong("retry.backoff.ms");
            boolean autoDowngradeTxnCommit = config.getBoolean("internal.auto.downgrade.txn.commit");
            transactionManager = new TransactionManager(logContext, transactionalId, transactionTimeoutMs, retryBackoffMs, this.apiVersions, autoDowngradeTxnCommit);
            if (transactionManager.isTransactional()) {
                this.log.info("Instantiated a transactional producer.");
            } else {
                this.log.info("Instantiated an idempotent producer.");
            }
        }

        return transactionManager;
    }

    private static int configureInflightRequests(ProducerConfig config) {
        if (config.idempotenceEnabled() && 5 < config.getInt("max.in.flight.requests.per.connection")) {
            throw new ConfigException("Must set max.in.flight.requests.per.connection to at most 5 to use the idempotent producer.");
        } else {
            return config.getInt("max.in.flight.requests.per.connection");
        }
    }

    private static short configureAcks(ProducerConfig config, Logger log) {
        boolean userConfiguredAcks = config.originals().containsKey("acks");
        short acks = Short.parseShort(config.getString("acks"));
        if (config.idempotenceEnabled()) {
            if (!userConfiguredAcks) {
                log.info("Overriding the default {} to all since idempotence is enabled.", "acks");
            } else if (acks != -1) {
                throw new ConfigException("Must set acks to all in order to use the idempotent producer. Otherwise we cannot guarantee idempotence.");
            }
        }

        return acks;
    }

    public void initTransactions() {
        this.throwIfNoTransactionManager();
        this.throwIfProducerClosed();
        TransactionalRequestResult result = this.transactionManager.initializeTransactions();
        this.sender.wakeup();
        result.await(this.maxBlockTimeMs, TimeUnit.MILLISECONDS);
    }

    public void beginTransaction() throws ProducerFencedException {
        this.throwIfNoTransactionManager();
        this.throwIfProducerClosed();
        this.transactionManager.beginTransaction();
    }

    public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) throws ProducerFencedException {
        this.sendOffsetsToTransaction(offsets, new ConsumerGroupMetadata(consumerGroupId));
    }

    public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, ConsumerGroupMetadata groupMetadata) throws ProducerFencedException {
        this.throwIfInvalidGroupMetadata(groupMetadata);
        this.throwIfNoTransactionManager();
        this.throwIfProducerClosed();
        TransactionalRequestResult result = this.transactionManager.sendOffsetsToTransaction(offsets, groupMetadata);
        this.sender.wakeup();
        result.await(this.maxBlockTimeMs, TimeUnit.MILLISECONDS);
    }

    public void commitTransaction() throws ProducerFencedException {
        this.throwIfNoTransactionManager();
        this.throwIfProducerClosed();
        TransactionalRequestResult result = this.transactionManager.beginCommit();
        this.sender.wakeup();
        result.await(this.maxBlockTimeMs, TimeUnit.MILLISECONDS);
    }

    public void abortTransaction() throws ProducerFencedException {
        this.throwIfNoTransactionManager();
        this.throwIfProducerClosed();
        this.log.info("Aborting incomplete transaction");
        TransactionalRequestResult result = this.transactionManager.beginAbort();
        this.sender.wakeup();
        result.await(this.maxBlockTimeMs, TimeUnit.MILLISECONDS);
    }

    public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
        return this.send(record, (Callback)null);
    }

    public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
        ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
        return this.doSend(interceptedRecord, callback);
    }

    private void throwIfProducerClosed() {
        if (this.sender == null || !this.sender.isRunning()) {
            throw new IllegalStateException("Cannot perform operation after producer has been closed");
        }
    }

    private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
        TopicPartition tp = null;

        try {
            this.throwIfProducerClosed();
            long nowMs = this.time.milliseconds();

            ClusterAndWaitTime clusterAndWaitTime;
            try {
                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;
            }

            nowMs += clusterAndWaitTime.waitedOnMetadataMs;
            long remainingWaitMs = Math.max(0L, this.maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);
            Cluster cluster = clusterAndWaitTime.cluster;

            byte[] serializedKey;
            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);
            }

            int partition = this.partition(record, serializedKey, serializedValue, cluster);
            tp = new TopicPartition(record.topic(), partition);
            this.setReadOnly(record.headers());
            Header[] headers = record.headers().toArray();
            int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(this.apiVersions.maxUsableProduceMagic(), this.compressionType, serializedKey, serializedValue, headers);
            this.ensureValidRecordSize(serializedSize);
            long timestamp = record.timestamp() == null ? nowMs : record.timestamp();
            if (this.log.isTraceEnabled()) {
                this.log.trace("Attempting to append record {} with callback {} to topic {} partition {}", new Object[]{record, callback, record.topic(), partition});
            }

            Callback interceptCallback = new InterceptorCallback(callback, this.interceptors, tp);
            if (this.transactionManager != null && this.transactionManager.isTransactional()) {
                this.transactionManager.failIfNotReadyForSend();
            }

            RecordAccumulator.RecordAppendResult result = this.accumulator.append(tp, timestamp, serializedKey, serializedValue, headers, interceptCallback, remainingWaitMs, true, nowMs);
            if (result.abortForNewBatch) {
                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 InterceptorCallback(callback, this.interceptors, tp);
                result = this.accumulator.append(tp, timestamp, serializedKey, serializedValue, headers, interceptCallback, remainingWaitMs, false, nowMs);
            }

            if (this.transactionManager != null && this.transactionManager.isTransactional()) {
                this.transactionManager.maybeAddPartitionToTransaction(tp);
            }

            if (result.batchIsFull || result.newBatchCreated) {
                this.log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), partition);
                this.sender.wakeup();
            }

            return result.future;
        } catch (ApiException var23) {
            this.log.debug("Exception occurred during message send:", var23);
            if (callback != null) {
                callback.onCompletion((RecordMetadata)null, var23);
            }

            this.errors.record();
            this.interceptors.onSendError(record, tp, var23);
            return new FutureFailure(var23);
        } catch (InterruptedException var24) {
            this.errors.record();
            this.interceptors.onSendError(record, tp, var24);
            throw new InterruptException(var24);
        } catch (KafkaException var25) {
            this.errors.record();
            this.interceptors.onSendError(record, tp, var25);
            throw var25;
        } catch (Exception var26) {
            this.interceptors.onSendError(record, tp, var26);
            throw var26;
        }
    }

    private void setReadOnly(Headers headers) {
        if (headers instanceof RecordHeaders) {
            ((RecordHeaders)headers).setReadOnly();
        }

    }

    private ClusterAndWaitTime waitOnMetadata(String topic, Integer partition, long nowMs, long maxWaitMs) throws InterruptedException {
        Cluster cluster = this.metadata.fetch();
        if (cluster.invalidTopics().contains(topic)) {
            throw new InvalidTopicException(topic);
        } else {
            this.metadata.add(topic, nowMs);
            Integer partitionsCount = cluster.partitionCountForTopic(topic);
            if (partitionsCount != null && (partition == null || partition < partitionsCount)) {
                return new ClusterAndWaitTime(cluster, 0L);
            } else {
                long remainingWaitMs = maxWaitMs;
                long elapsed = 0L;

                do {
                    do {
                        if (partition != null) {
                            this.log.trace("Requesting metadata update for partition {} of topic {}.", partition, topic);
                        } else {
                            this.log.trace("Requesting metadata update for topic {}.", topic);
                        }

                        this.metadata.add(topic, nowMs + elapsed);
                        int version = this.metadata.requestUpdateForTopic(topic);
                        this.sender.wakeup();

                        try {
                            this.metadata.awaitUpdate(version, remainingWaitMs);
                        } catch (TimeoutException var15) {
                            throw new TimeoutException(String.format("Topic %s not present in metadata after %d ms.", topic, maxWaitMs));
                        }

                        cluster = this.metadata.fetch();
                        elapsed = this.time.milliseconds() - nowMs;
                        if (elapsed >= maxWaitMs) {
                            throw new TimeoutException(partitionsCount == null ? String.format("Topic %s not present in metadata after %d ms.", topic, maxWaitMs) : String.format("Partition %d of topic %s with partition count %d is not present in metadata after %d ms.", partition, topic, partitionsCount, maxWaitMs));
                        }

                        this.metadata.maybeThrowExceptionForTopic(topic);
                        remainingWaitMs = maxWaitMs - elapsed;
                        partitionsCount = cluster.partitionCountForTopic(topic);
                    } while(partitionsCount == null);
                } while(partition != null && partition >= partitionsCount);

                return new ClusterAndWaitTime(cluster, elapsed);
            }
        }
    }

    private void ensureValidRecordSize(int size) {
        if (size > this.maxRequestSize) {
            throw new RecordTooLargeException("The message is " + size + " bytes when serialized which is larger than " + this.maxRequestSize + ", which is the value of the " + "max.request.size" + " configuration.");
        } else if ((long)size > this.totalMemorySize) {
            throw new RecordTooLargeException("The message is " + size + " bytes when serialized which is larger than the total memory buffer you have configured with the " + "buffer.memory" + " configuration.");
        }
    }

    public void flush() {
        this.log.trace("Flushing accumulated records in producer.");
        this.accumulator.beginFlush();
        this.sender.wakeup();

        try {
            this.accumulator.awaitFlushCompletion();
        } catch (InterruptedException var2) {
            throw new InterruptException("Flush interrupted.", var2);
        }
    }

    public List<PartitionInfo> partitionsFor(String topic) {
        Objects.requireNonNull(topic, "topic cannot be null");

        try {
            return this.waitOnMetadata(topic, (Integer)null, this.time.milliseconds(), this.maxBlockTimeMs).cluster.partitionsForTopic(topic);
        } catch (InterruptedException var3) {
            throw new InterruptException(var3);
        }
    }

    public Map<MetricName, ? extends Metric> metrics() {
        return Collections.unmodifiableMap(this.metrics.metrics());
    }

    public void close() {
        this.close(Duration.ofMillis(Long.MAX_VALUE));
    }

    public void close(Duration timeout) {
        this.close(timeout, false);
    }

    private void close(Duration timeout, boolean swallowException) {
        long timeoutMs = timeout.toMillis();
        if (timeoutMs < 0L) {
            throw new IllegalArgumentException("The timeout cannot be negative.");
        } else {
            this.log.info("Closing the Kafka producer with timeoutMillis = {} ms.", timeoutMs);
            AtomicReference<Throwable> firstException = new AtomicReference();
            boolean invokedFromCallback = Thread.currentThread() == this.ioThread;
            if (timeoutMs > 0L) {
                if (invokedFromCallback) {
                    this.log.warn("Overriding close timeout {} ms to 0 ms in order to prevent useless blocking due to self-join. This means you have incorrectly invoked close with a non-zero timeout from the producer call-back.", timeoutMs);
                } else {
                    if (this.sender != null) {
                        this.sender.initiateClose();
                    }

                    if (this.ioThread != null) {
                        try {
                            this.ioThread.join(timeoutMs);
                        } catch (InterruptedException var9) {
                            firstException.compareAndSet((Object)null, new InterruptException(var9));
                            this.log.error("Interrupted while joining ioThread", var9);
                        }
                    }
                }
            }

            if (this.sender != null && this.ioThread != null && this.ioThread.isAlive()) {
                this.log.info("Proceeding to force close the producer since pending requests could not be completed within timeout {} ms.", timeoutMs);
                this.sender.forceClose();
                if (!invokedFromCallback) {
                    try {
                        this.ioThread.join();
                    } catch (InterruptedException var8) {
                        firstException.compareAndSet((Object)null, new InterruptException(var8));
                    }
                }
            }

            Utils.closeQuietly(this.interceptors, "producer interceptors", firstException);
            Utils.closeQuietly(this.metrics, "producer metrics", firstException);
            Utils.closeQuietly(this.keySerializer, "producer keySerializer", firstException);
            Utils.closeQuietly(this.valueSerializer, "producer valueSerializer", firstException);
            Utils.closeQuietly(this.partitioner, "producer partitioner", firstException);
            AppInfoParser.unregisterAppInfo("kafka.producer", this.clientId, this.metrics);
            Throwable exception = (Throwable)firstException.get();
            if (exception != null && !swallowException) {
                if (exception instanceof InterruptException) {
                    throw (InterruptException)exception;
                } else {
                    throw new KafkaException("Failed to close kafka producer", exception);
                }
            } else {
                this.log.debug("Kafka producer has been closed");
            }
        }
    }

    private ClusterResourceListeners configureClusterResourceListeners(Serializer<K> keySerializer, Serializer<V> valueSerializer, List<?>... candidateLists) {
        ClusterResourceListeners clusterResourceListeners = new ClusterResourceListeners();
        List[] var5 = candidateLists;
        int var6 = candidateLists.length;

        for(int var7 = 0; var7 < var6; ++var7) {
            List<?> candidateList = var5[var7];
            clusterResourceListeners.maybeAddAll(candidateList);
        }

        clusterResourceListeners.maybeAdd(keySerializer);
        clusterResourceListeners.maybeAdd(valueSerializer);
        return clusterResourceListeners;
    }

    private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
        Integer partition = record.partition();
        return partition != null ? partition : this.partitioner.partition(record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
    }

    private void throwIfInvalidGroupMetadata(ConsumerGroupMetadata groupMetadata) {
        if (groupMetadata == null) {
            throw new IllegalArgumentException("Consumer group metadata could not be null");
        } else if (groupMetadata.generationId() > 0 && "".equals(groupMetadata.memberId())) {
            throw new IllegalArgumentException("Passed in group metadata " + groupMetadata + " has generationId > 0 but member.id ");
        }
    }

    private void throwIfNoTransactionManager() {
        if (this.transactionManager == null) {
            throw new IllegalStateException("Cannot use transactional methods without enabling transactions by setting the transactional.id configuration property");
        }
    }

    String getClientId() {
        return this.clientId;
    }

    private static class InterceptorCallback<K, V> implements Callback {
        private final Callback userCallback;
        private final ProducerInterceptors<K, V> interceptors;
        private final TopicPartition tp;

        private InterceptorCallback(Callback userCallback, ProducerInterceptors<K, V> interceptors, TopicPartition tp) {
            this.userCallback = userCallback;
            this.interceptors = interceptors;
            this.tp = tp;
        }

        public void onCompletion(RecordMetadata metadata, Exception exception) {
            metadata = metadata != null ? metadata : new RecordMetadata(this.tp, -1L, -1L, -1L, -1L, -1, -1);
            this.interceptors.onAcknowledgement(metadata, exception);
            if (this.userCallback != null) {
                this.userCallback.onCompletion(metadata, exception);
            }

        }
    }

    private static class FutureFailure implements Future<RecordMetadata> {
        private final ExecutionException exception;

        public FutureFailure(Exception exception) {
            this.exception = new ExecutionException(exception);
        }

        public boolean cancel(boolean interrupt) {
            return false;
        }

        public RecordMetadata get() throws ExecutionException {
            throw this.exception;
        }

        public RecordMetadata get(long timeout, TimeUnit unit) throws ExecutionException {
            throw this.exception;
        }

        public boolean isCancelled() {
            return false;
        }

        public boolean isDone() {
            return true;
        }
    }

    private static class ClusterAndWaitTime {
        final Cluster cluster;
        final long waitedOnMetadataMs;

        ClusterAndWaitTime(Cluster cluster, long waitedOnMetadataMs) {
            this.cluster = cluster;
            this.waitedOnMetadataMs = waitedOnMetadataMs;
        }
    }
}