pinpoint-apm/pinpoint · error

onFailure

Error message

{} onFailure:{}

What it means

KafkaCallbacks.loggingCallback returns a BiConsumer attached to Kafka send futures; when the KafkaTemplate.send future completes with a Throwable, the callback logs this warning containing the callback name and throwable message. It indicates a Kafka produce failure (message may not have been persisted to the topic).

Solutions

1. Read the logged throwable for the exact Kafka error (TimeoutException, NotLeaderForPartition, RecordTooLargeException, etc.).
2. Verify bootstrap.servers connectivity and topic existence/ACLs from the collector host.
3. Increase delivery.timeout.ms / request.timeout.ms or reduce batch pressure under load.
4. Add a producer failure metric/retry policy or dead-letter path for lost records.

Example fix

// before: failure only logged, data silently lost
kafkaTemplate.send(topic, record, KafkaCallbacks.loggingCallback("span", log));
// after: handle failure explicitly
kafkaTemplate.send(topic, record)
    .whenComplete(KafkaCallbacks.loggingCallback("span", log));
// plus producer config: delivery.timeout.ms=120000, retries=10

// pre-check before send
try (AdminClient admin = AdminClient.create(props)) {
    if (!admin.describeTopics(Collections.singletonList(topic)).allTopicNames().get().containsKey(topic)) {
        throw new IllegalStateException("topic missing: " + topic);
    }
}

kafkaTemplate.send(topic, key, value)
    .whenComplete((result, throwable) -> {
        if (throwable != null) {
            logger.warn("send onFailure:{}", throwable.getMessage(), throwable);
            // deadLetter(topic, key, value, throwable); // persist for replay
        }
    });

When it happens

Trigger: The ListenableFuture/CompletableFuture from a Kafka producer send completes exceptionally — broker unreachable, record too large, serialization failure, timeout, or topic authorization failure.

Common situations: Kafka broker down or DNS/service name wrong in pinot-kafka config; acks/timeout settings too aggressive; message exceeding max.request.size; topic missing or ACLs deny produce.

Understand the failure class

Background: "API request failed": what wrapped HTTP errors from external APIs mean and how to find the real cause — this error's family across 29 libraries.

AI-assisted analysis of pinpoint-apm/pinpoint@744c3d3075 (2026-09-07). Data as JSON: /api/errors/59d0e023a30b8953. Report an issue: GitHub.

Appendix: source

Thrown at pinot/pinot-kafka/src/main/java/com/navercorp/pinpoint/pinot/kafka/util/KafkaCallbacks.java:15

package com.navercorp.pinpoint.pinot.kafka.util;

import org.apache.logging.log4j.Logger;
import org.springframework.kafka.support.SendResult;

import java.util.function.BiConsumer;

public final class KafkaCallbacks {

    public static <T> BiConsumer<SendResult<String, T>, Throwable> loggingCallback(String name, Logger logger) {
        return new BiConsumer<>() {
            @Override
            public void accept(SendResult<String, T> result, Throwable throwable) {
                if (throwable != null) {
                    logger.warn("{} onFailure:{}", name, throwable.getMessage(), throwable);
                } else {
                    if (logger.isDebugEnabled()) {
                        logger.debug("{} onSuccess:{}", name, result);
                    }
                }
            }
        };
    }
}

View on GitHub (pinned to 744c3d3075)