{"record":{"id":"27c52080b200eee3","repo":"apache/beam","slug":"kafkawriter-failed-to-send-d-records-since-last-report","errorCode":null,"errorMessage":"KafkaWriter : failed to send %d records (since last report)","messagePattern":"KafkaWriter : failed to send (.+?) records \\(since last report\\)","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriter.java","lineNumber":189,"sourceCode":"  WriteRecords<K, V> getSpec() {\n    return spec;\n  }\n\n  private synchronized void checkForFailures() throws IOException {\n    if (numSendFailures == 0) {\n      return;\n    }\n\n    String msg =\n        String.format(\n            \"KafkaWriter : failed to send %d records (since last report)\", numSendFailures);\n\n    Exception e = sendException;\n    sendException = null;\n    numSendFailures = 0;\n\n    LOG.warn(\"{}\", msg);\n    throw new IOException(msg, e);\n  }\n\n  private class SendCallback implements Callback {\n    @Override\n    public void onCompletion(RecordMetadata metadata, Exception exception) {\n      if (exception == null) {\n        return;\n      }\n\n      synchronized (KafkaWriter.this) {\n        if (sendException == null) {\n          sendException = exception;\n        }\n        numSendFailures++;\n        // don't log exception stacktrace here, exception will be propagated up.\n        LOG.warn(\"send failed\", exception);\n      }\n    }","sourceCodeStart":171,"sourceCodeEnd":207,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriter.java#L171-L207","documentation":"KafkaWriter.checkForFailures is the error surfacing point for asynchronous sends: failed sends are counted in the SendCallback, and at bundle checkpoints (processElement/finishBundle) this method converts the accumulated failure count and last exception into an IOException. It means the Kafka producer failed to deliver records since the last report (broker unavailable, timeouts, record too large, etc.).","triggerScenarios":"Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriter.java:189 when the library encounters an invalid state.","commonSituations":"See trigger scenarios.","solutions":["Check the wrapped sendException for the root cause (broker connectivity, timeouts, message size, auth).","Verify Kafka broker availability, bootstrap servers config, and topic settings.","Retry the pipeline/failed bundle; consider producer retries (retries, acks, delivery.timeout.ms) tuning."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":null,"preventionTips":[],"tags":[],"backgroundTag":null,"analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}