apache/beam · error · IOException

KafkaWriter : failed to send

Error message

KafkaWriter : failed to send %d records (since last report)

What it means

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.).

Solutions

1. Check the wrapped sendException for the root cause (broker connectivity, timeouts, message size, auth).
2. Verify Kafka broker availability, bootstrap servers config, and topic settings.
3. Retry the pipeline/failed bundle; consider producer retries (retries, acks, delivery.timeout.ms) tuning.

When it happens

Trigger: 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.

Common situations: See trigger scenarios.

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/27c52080b200eee3. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriter.java:189

  WriteRecords<K, V> getSpec() {
    return spec;
  }

  private synchronized void checkForFailures() throws IOException {
    if (numSendFailures == 0) {
      return;
    }

    String msg =
        String.format(
            "KafkaWriter : failed to send %d records (since last report)", numSendFailures);

    Exception e = sendException;
    sendException = null;
    numSendFailures = 0;

    LOG.warn("{}", msg);
    throw new IOException(msg, e);
  }

  private class SendCallback implements Callback {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
      if (exception == null) {
        return;
      }

      synchronized (KafkaWriter.this) {
        if (sendException == null) {
          sendException = exception;
        }
        numSendFailures++;
        // don't log exception stacktrace here, exception will be propagated up.
        LOG.warn("send failed", exception);
      }
    }

View on GitHub (pinned to 12126d8942)