{"record":{"id":"59d0e023a30b8953","repo":"pinpoint-apm/pinpoint","slug":"onfailure","errorCode":null,"errorMessage":"{} onFailure:{}","messagePattern":"(.+?) onFailure:(.+?)","errorType":"console","errorClass":null,"httpStatus":null,"severity":"error","filePath":"pinot/pinot-kafka/src/main/java/com/navercorp/pinpoint/pinot/kafka/util/KafkaCallbacks.java","lineNumber":15,"sourceCode":"package com.navercorp.pinpoint.pinot.kafka.util;\n\nimport org.apache.logging.log4j.Logger;\nimport org.springframework.kafka.support.SendResult;\n\nimport java.util.function.BiConsumer;\n\npublic final class KafkaCallbacks {\n\n    public static <T> BiConsumer<SendResult<String, T>, Throwable> loggingCallback(String name, Logger logger) {\n        return new BiConsumer<>() {\n            @Override\n            public void accept(SendResult<String, T> result, Throwable throwable) {\n                if (throwable != null) {\n                    logger.warn(\"{} onFailure:{}\", name, throwable.getMessage(), throwable);\n                } else {\n                    if (logger.isDebugEnabled()) {\n                        logger.debug(\"{} onSuccess:{}\", name, result);\n                    }\n                }\n            }\n        };\n    }\n}\n","sourceCodeStart":1,"sourceCodeEnd":25,"githubUrl":"https://github.com/pinpoint-apm/pinpoint/blob/744c3d3075e595656abb1ae331ad2c0e4c9eb996/pinot/pinot-kafka/src/main/java/com/navercorp/pinpoint/pinot/kafka/util/KafkaCallbacks.java#L1-L25","documentation":"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).","triggerScenarios":"The ListenableFuture/CompletableFuture from a Kafka producer send completes exceptionally — broker unreachable, record too large, serialization failure, timeout, or topic authorization failure.","commonSituations":"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.","solutions":["Read the logged throwable for the exact Kafka error (TimeoutException, NotLeaderForPartition, RecordTooLargeException, etc.).","Verify bootstrap.servers connectivity and topic existence/ACLs from the collector host.","Increase delivery.timeout.ms / request.timeout.ms or reduce batch pressure under load.","Add a producer failure metric/retry policy or dead-letter path for lost records."],"exampleFix":"// before: failure only logged, data silently lost\nkafkaTemplate.send(topic, record, KafkaCallbacks.loggingCallback(\"span\", log));\n// after: handle failure explicitly\nkafkaTemplate.send(topic, record)\n    .whenComplete(KafkaCallbacks.loggingCallback(\"span\", log));\n// plus producer config: delivery.timeout.ms=120000, retries=10","handlingStrategy":"try-catch","validationCode":"// pre-check before send\ntry (AdminClient admin = AdminClient.create(props)) {\n    if (!admin.describeTopics(Collections.singletonList(topic)).allTopicNames().get().containsKey(topic)) {\n        throw new IllegalStateException(\"topic missing: \" + topic);\n    }\n}","typeGuard":null,"tryCatchPattern":"kafkaTemplate.send(topic, key, value)\n    .whenComplete((result, throwable) -> {\n        if (throwable != null) {\n            logger.warn(\"send onFailure:{}\", throwable.getMessage(), throwable);\n            // deadLetter(topic, key, value, throwable); // persist for replay\n        }\n    });","preventionTips":["Verify bootstrap.servers, topic existence, and produce ACLs before deploying.","Set sane producer timeouts (delivery.timeout.ms, request.timeout.ms) and retries.","Check message sizes against max.request.size to avoid RecordTooLargeException.","Monitor Kafka producer error metrics; the callback only logs, it does not recover."],"tags":["kafka","producer","async-callback","send-failure"],"backgroundTag":"api-request-failed","analyzedSha":"744c3d3075e595656abb1ae331ad2c0e4c9eb996","analyzedAt":"2026-09-07T18:48:45.289Z","contentChangedAt":"2026-09-07T18:48:45.289Z","schemaVersion":2},"datasetVersion":"2026-09-17T15:17:12.973Z"}