Kafka 生产者客户端几种异常 Case 详解

ERROR Error executing user-provided callback on message for topic-partition 'Topic1-0' (org.apache.kafka.clients.producer.internals.ProducerBatch)
通常还会有具体的异常栈信息

    private void completeFutureAndFireCallbacks(long baseOffset, long logAppendTime, RuntimeException exception) {        // Set the future before invoking the callbacks as we rely on its state for the `onCompletion` call        produceFuture.set(baseOffset, logAppendTime, exception);
        // execute callbacks        for (Thunk thunk : thunks) {            try {                if (exception == null) {                    RecordMetadata metadata = thunk.future.value();                    if (thunk.callback != null)                        thunk.callback.onCompletion(metadata, null);                } else {                    if (thunk.callback != null)                        thunk.callback.onCompletion(null, exception);                }            } catch (Exception e) {                log.error("Error executing user-provided callback on message for topic-partition '{}'", topicPartition, e);            }        }
        produceFuture.done();    }

Failed to allocate memory within the configured max blocking time 60000 ms.

public ByteBuffer allocate(int size, long maxTimeToBlockMs) throws InterruptedException {        
        try {            // check if we have a free buffer of the right size pooled            if (size == poolableSize && !this.free.isEmpty()){                              return this.free.pollFirst();            }
            // now check if the request is immediately satisfiable with the            // memory on hand or if we need to block            int freeListSize = freeSize() * this.poolableSize;            if (this.nonPooledAvailableMemory + freeListSize >= size) {                //省略...            } else {                // 当内存不够的时候将会被阻塞阻塞                int accumulated = 0;                Condition moreMemory = this.lock.newCondition();                try {                    long remainingTimeToBlockNs = TimeUnit.MILLISECONDS.toNanos(maxTimeToBlockMs);                    this.waiters.addLast(moreMemory);                    // loop over and over until we have a buffer or have reserved                    // enough memory to allocate one                    // 当内存不够,则会阻塞,当有内存释放的时候会唤醒阻塞,继续内存分配                    // 但是释放的内存不一定满足当前需要的内存size，则继续阻塞，等到下一次的内存继续释放,循环这个过程知道内存足够分配。                    // 阻塞的最大时长maxTimeToBlockMs，注意：就算循环了多次唤醒分配,这个时候是从第一次开始算的，也就是累积时间超过这个时间就会超时                    while (accumulated < size) {                        long startWaitNs = time.nanoseconds();                        long timeNs;                        boolean waitingTimeElapsed;                        try {                            System.out.println("allocate开始阻塞,等待内存释放,剩余等待时间：="+remainingTimeToBlockNs);                            waitingTimeElapsed = !moreMemory.await(remainingTimeToBlockNs, TimeUnit.NANOSECONDS);                        } finally {                            long endWaitNs = time.nanoseconds();                            timeNs = Math.max(0L, endWaitNs - startWaitNs);                            recordWaitTime(timeNs);                        }
                        if (this.closed)                            throw new KafkaException("Producer closed while allocating memory");
                        if (waitingTimeElapsed) {                            this.metrics.sensor("buffer-exhausted-records").record();                            throw new BufferExhaustedException("Failed to allocate memory within the configured max blocking time " + maxTimeToBlockMs + " ms.");                        }
                        remainingTimeToBlockNs -= timeNs;
                        // check if we can satisfy this request from the free list,                        // otherwise allocate memory                        if (accumulated == 0 && size == this.poolableSize && !this.free.isEmpty()) {                            // just grab a buffer from the free list                            // 从内存池中获取第一个bytebuffer返回                            buffer = this.free.pollFirst();                            accumulated = size;                            System.out.println("allocate 被唤醒,将bufferPool第一个pool返回并分配：size="+size);
                        } else {                                //省略....                        }                    }
                } finally {                    //省略....                }            }        } finally {                  //省略....        }    }

[2022-04-25 13:42:23,457] WARN [Producer clientId=console-producer] Got error produce response with correlation id 47 on topic-partition Topic4-0, retrying (2 attempts left). Error: NETWORK_EXCEPTION (org.apache.kafka.clients.producer.internals.Sender)[2022-04-25 13:42:54,604] WARN [Producer clientId=console-producer] Received invalid metadata error in produce request on partition Topic4-0 due to org.apache.kafka.common.errors.NetworkException: The server disconnected before a response was received.. Going to request metadata update now (org.apache.kafka.clients.producer.internals.Sender)

   if (error.exception() instanceof InvalidMetadataException) {                if (error.exception() instanceof UnknownTopicOrPartitionException) {                    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 {                    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());                }                metadata.requestUpdate();            }

Received invalid metadata error in produce request on partition Topic4-0 due to org.apache.kafka.common.errors.NetworkException: The server disconnected before a response was received.. Going to request metadata update now (org.apache.kafka.clients.producer.internals.Sender)

Got error produce response with correlation id 47 on topic-partition Topic4-0, retrying (2 attempts left). Error: NETWORK_EXCEPTION 

org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for Topic4-0:4010 ms has passed since batch creation

org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for Topic4-0:4010 ms has passed since batch creation

        List<ProducerBatch> expiredInflightBatches = getExpiredInflightBatches(now);        List<ProducerBatch> expiredBatches = this.accumulator.expiredBatches(now);        expiredBatches.addAll(expiredInflightBatches);
        if (!expiredBatches.isEmpty())            log.trace("Expired {} batches in accumulator", expiredBatches.size());        for (ProducerBatch expiredBatch : expiredBatches) {            String errorMessage = "Expiring " + expiredBatch.recordCount + " record(s) for " + expiredBatch.topicPartition                + ":" + (now - expiredBatch.createdMs) + " ms has passed since batch creation";            failBatch(expiredBatch, -1, NO_TIMESTAMP, new TimeoutException(errorMessage), false);            if (transactionManager != null && expiredBatch.inRetry()) {                // This ensures that no new batches are drained until the current in flight batches are fully resolved.                transactionManager.markSequenceUnresolved(expiredBatch);            }        }