dotnetcore/CAP · error · PublisherSentFailedException

kafka message persisted failed!

Error message

kafka message persisted failed!

What it means

Operational (not framework) failure logged in Kafka transport SendAsync: ProduceAsync returned a PersistenceStatus other than Persisted/PossiblyPersisted (e.g. NotPersisted), so the producer could not confirm the broker accepted the message. Typically caused by broker unavailability, timeouts, message size limits, or ack/replication failures.

Solutions

1. Verify Kafka broker reachability (bootstrap servers, network, TLS) and cluster health
2. Increase delivery timeout / retry settings (MessageTimeoutMs, MessageSendMaxRetries) on the producer
3. Check topic exists, is not throttled, and max.message.bytes is not exceeded
4. Inspect the logged result status and broker logs to distinguish definite loss from ambiguous delivery

When it happens

Trigger: Thrown at src/DotNetCore.CAP.Kafka/ITransport.Kafka.cs:61 when the library encounters an invalid state.

Common situations: See trigger scenarios.

AI-assisted analysis of dotnetcore/CAP@e52b8508e5 (2026-09-14). Data as JSON: /api/errors/adee746a92807f7b. Report an issue: GitHub.

Appendix: source

Thrown at src/DotNetCore.CAP.Kafka/ITransport.Kafka.cs:61

            var result = await producer.ProduceAsync(message.GetName(), new Message<string, byte[]>
            {
                Headers = headers,
                Key = message.Headers.TryGetValue(KafkaHeaders.KafkaKey, out var kafkaMessageKey) &&
                      !string.IsNullOrEmpty(kafkaMessageKey)
                    ? kafkaMessageKey
                    : message.GetId(),
                Value = message.Body.ToArray()!
            });

            if (result.Status is PersistenceStatus.Persisted or PersistenceStatus.PossiblyPersisted)
            {
                _logger.LogDebug($"kafka topic message [{message.GetName()}] has been published.");

                return OperateResult.Success;
            }

            throw new PublisherSentFailedException("kafka message persisted failed!");
        }
        catch (Exception ex)
        {
            var warpEx = new PublisherSentFailedException(ex.Message, ex);

            return OperateResult.Failed(warpEx);
        }
        finally
        {
            _connectionPool.Return(producer);
        }
    }
}

View on GitHub (pinned to e52b8508e5)