public class KafkaProducerConfigurationBuilder : KafkaClientConfigurationBuilder<KafkaProducerConfiguration, ProducerConfig, KafkaProducerConfigurationBuilder>

Constructors

KafkaProducerConfigurationBuilder(IServiceProvider)

Initializes a new instance of the KafkaProducerConfigurationBuilder class.

Declaration

public KafkaProducerConfigurationBuilder(IServiceProvider serviceProvider)

Parameters

Type	Name	Description
IServiceProvider	serviceProvider	The IServiceProvider.

Properties

This

Gets this instance.

Declaration

protected override KafkaProducerConfigurationBuilder This { get; }

Property Value

Type	Description
KafkaProducerConfigurationBuilder

Overrides

KafkaClientConfigurationBuilder<KafkaProducerConfiguration, ProducerConfig, KafkaProducerConfigurationBuilder>.This

Remarks

This is necessary to work around casting in the base classes.

Methods

Build()

Builds the KafkaProducerConfiguration instance.

Declaration

public KafkaProducerConfiguration Build()

Returns

Type	Description
KafkaProducerConfiguration	The KafkaProducerConfiguration.

BuildCore()

Builds the configuration.

Declaration

protected override KafkaProducerConfiguration BuildCore()

Returns

Type	Description
KafkaProducerConfiguration	The configuration.

Overrides

KafkaClientConfigurationBuilder<KafkaProducerConfiguration, ProducerConfig, KafkaProducerConfigurationBuilder>.BuildCore()

DisableDeliveryReports()

Disable notification of delivery reports. This will enable "fire and forget" semantics and give a small boost in performance.

Declaration

public KafkaProducerConfigurationBuilder DisableDeliveryReports()

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

DisableDisposeOnException()

Specifies tjat the producer don't have to be disposed and recreated if a Confluent.Kafka.KafkaException is thrown.

Declaration

public KafkaProducerConfigurationBuilder DisableDisposeOnException()

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

DisableGaplessGuarantee()

Disables the gapless guarantee.

Declaration

public KafkaProducerConfigurationBuilder DisableGaplessGuarantee()

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

DisableIdempotence()

The producer will not ensure that messages are successfully produced exactly once and in the original produce order.

Declaration

public KafkaProducerConfigurationBuilder DisableIdempotence()

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

DisableTransactions()

Disables the Kafka transactions. The producer is limited to idempotent delivery (see EnableIdempotence()).

Declaration

public KafkaProducerConfigurationBuilder DisableTransactions()

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

DisposeOnException()

Specifies that the producer has to be disposed and recreated if a Confluent.Kafka.KafkaException is thrown. This is the default.

Declaration

public KafkaProducerConfigurationBuilder DisposeOnException()

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

EnableDeliveryReports()

Enable notification of delivery reports.

Declaration

public KafkaProducerConfigurationBuilder EnableDeliveryReports()

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

EnableGaplessGuarantee()

Specifies that an error that could result in a gap in the produced message series when a batch of messages fails, must raise a fatal error (ERR_GAPLESS_GUARANTEE) and stop the producer. Messages failing due to MessageTimeoutMs are not covered by this guarantee. Requires EnableIdempotence()=true.

Declaration

public KafkaProducerConfigurationBuilder EnableGaplessGuarantee()

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

EnableIdempotence()

The producer will ensure that messages are successfully produced exactly once and in the original produce order. The following configuration properties are adjusted automatically (if not modified by the user) when idempotence is enabled: max.in.flight.requests.per.connection=5 (must be less than or equal to 5), retries=INT32_MAX (must be greater than 0), acks=all, queuing.strategy=fifo. Producer instantiation will fail if user-supplied configuration is incompatible.

Declaration

public KafkaProducerConfigurationBuilder EnableIdempotence()

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

EnableTransactions(string)

Enables the Kafka transactions and sets the identifier to be used to identify the same transactional producer instance across process restarts. This allows the producer to guarantee that transactions corresponding to earlier instances of the same producer have been finalized prior to starting any new transaction, and that any zombie instances are fenced off. Requires broker version >= 0.11.0.

Declaration

public KafkaProducerConfigurationBuilder EnableTransactions(string transactionalId)

Parameters

Type	Name	Description
string	transactionalId	The identifier to be used to identify the same transactional producer instance across process restarts.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

IgnoreIfNotAcknowledged()

Specifies that no exception has be thrown by the producer if the persistence is not acknowledge by the broker.

Declaration

public KafkaProducerConfigurationBuilder IgnoreIfNotAcknowledged()

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

Produce(Action<KafkaProducerEndpointConfigurationBuilder<object>>)

Adds a producer endpoint, which is a topic or partition and its related configuration (serializer, etc.).

Declaration

public KafkaProducerConfigurationBuilder Produce(Action<KafkaProducerEndpointConfigurationBuilder<object>> configurationBuilderAction)

Parameters

Type	Name	Description
Action<KafkaProducerEndpointConfigurationBuilder<object>>	configurationBuilderAction	An Action that takes the KafkaProducerConfigurationBuilder and configures it.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

Produce(string?, Action<KafkaProducerEndpointConfigurationBuilder<object>>)

Adds a producer endpoint, which is a topic or partition and its related configuration (serializer, etc.).

Declaration

public KafkaProducerConfigurationBuilder Produce(string? name, Action<KafkaProducerEndpointConfigurationBuilder<object>> configurationBuilderAction)

Parameters

Type	Name	Description
string	name	The name is used to guarantee that a duplicated configuration is discarded and is also displayed in the logs. By default, the name will be generated concatenating the topic name and the message type.
Action<KafkaProducerEndpointConfigurationBuilder<object>>	configurationBuilderAction	An Action that takes the KafkaProducerConfigurationBuilder and configures it.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

Produce<TMessage>(Action<KafkaProducerEndpointConfigurationBuilder<TMessage>>)

Adds a producer endpoint, which is a topic or partition and its related configuration (serializer, etc.).

Declaration

public KafkaProducerConfigurationBuilder Produce<TMessage>(Action<KafkaProducerEndpointConfigurationBuilder<TMessage>> configurationBuilderAction) where TMessage : class

Parameters

Type	Name	Description
Action<KafkaProducerEndpointConfigurationBuilder<TMessage>>	configurationBuilderAction	An Action that takes the KafkaProducerConfigurationBuilder and configures it.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

Type Parameters

Name	Description
TMessage	The type (or base type) of the messages being produced. This is used to setup the serializer and will determine the type of the message parameter in the nested configuration functions.

Produce<TMessage>(string?, Action<KafkaProducerEndpointConfigurationBuilder<TMessage>>)

Adds a producer endpoint, which is a topic or partition and its related configuration (serializer, etc.).

Declaration

public KafkaProducerConfigurationBuilder Produce<TMessage>(string? name, Action<KafkaProducerEndpointConfigurationBuilder<TMessage>> configurationBuilderAction) where TMessage : class

Parameters

Type	Name	Description
string	name	The name is used to guarantee that a duplicated configuration is discarded and is also displayed in the logs. By default, the name will be generated concatenating the topic name and the message type.
Action<KafkaProducerEndpointConfigurationBuilder<TMessage>>	configurationBuilderAction	An Action that takes the KafkaProducerConfigurationBuilder and configures it.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

Type Parameters

Name	Description
TMessage	The type (or base type) of the messages being produced. This is used to set up the serializer and will determine the type of the message parameter in the nested configuration functions.

ThrowIfNotAcknowledged()

Specifies that an exception must be thrown by the producer if the persistence is not acknowledge by the broker. This is the default.

Declaration

public KafkaProducerConfigurationBuilder ThrowIfNotAcknowledged()

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithBatchNumMessages(int?)

Sets the maximum number of messages batched in one message set. The total message set size is also limited by BatchSize and MessageMaxBytes.

Declaration

public KafkaProducerConfigurationBuilder WithBatchNumMessages(int? batchNumMessages)

Parameters

Type	Name	Description
int?	batchNumMessages	The maximum number of messages batched in one message set.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithBatchSize(int?)

Sets the maximum size (in bytes) of all messages batched in one message set, including the protocol framing overhead. This limit is applied after the first message has been added to the batch, regardless of the first message size, this is to ensure that messages that exceed the BatchSize are still produced. The total message set size is also limited by BatchNumMessages and MessageMaxBytes.

Declaration

public KafkaProducerConfigurationBuilder WithBatchSize(int? batchSize)

Parameters

Type	Name	Description
int?	batchSize	The maximum size of all messages batched in one message set.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithCompressionLevel(int?)

Sets the compression level parameter for the algorithm selected by configuration property Confluent.Kafka.CompressionType. Higher values will result in better compression at the cost of higher CPU usage. Usable range is algorithm-dependent: [0-9] for gzip, [0-12] for lz4, only 0 for snappy. -1 = codec-dependent default compression level.

Declaration

public KafkaProducerConfigurationBuilder WithCompressionLevel(int? compressionLevel)

Parameters

Type	Name	Description
int?	compressionLevel	The compression level parameter for the algorithm selected by configuration property Confluent.Kafka.CompressionType.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithCompressionType(CompressionType?)

Sets the compression codec to be used to compress message sets. This is the default value for all topics, may be overridden by the topic configuration property compression.codec.

Declaration

public KafkaProducerConfigurationBuilder WithCompressionType(CompressionType? compressionType)

Parameters

Type	Name	Description
CompressionType?	compressionType	The compression codec to be used to compress message sets.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithFlushTimeout(TimeSpan)

Specifies the flush operation timeout. The default is 30 seconds.

Declaration

public KafkaProducerConfigurationBuilder WithFlushTimeout(TimeSpan timeout)

Parameters

Type	Name	Description
TimeSpan	timeout	The flush operation timeout.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithLingerMs(double?)

Sets the delay in milliseconds to wait for messages in the producer queue to accumulate before constructing message batches to transmit to brokers. A higher value allows larger and more effective (less overhead, improved compression) batches of messages to accumulate at the expense of increased message delivery latency.

Declaration

public KafkaProducerConfigurationBuilder WithLingerMs(double? lingerMs)

Parameters

Type	Name	Description
double?	lingerMs	The delay to wait for messages in the producer queue to accumulate before constructing message batches.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithMessageSendMaxRetries(int?)

Sets how many times to retry sending a failing message.
Note: retrying may cause reordering unless EnableIdempotence() is set to true.

Declaration

public KafkaProducerConfigurationBuilder WithMessageSendMaxRetries(int? messageSendMaxRetries)

Parameters

Type	Name	Description
int?	messageSendMaxRetries	How many times to retry sending a failing message.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithMessageTimeoutMs(int?)

Sets the local message timeout (in milliseconds). This value is only enforced locally and limits the time a produced message waits for successful delivery. A time of 0 is infinite. This is the maximum time to deliver a message (including retries) and a delivery error will occur when either the retry count or the message timeout are exceeded. The message timeout is automatically adjusted to TransactionTimeoutMs if TransactionalId is set.

Declaration

public KafkaProducerConfigurationBuilder WithMessageTimeoutMs(int? messageTimeoutMs)

Parameters

Type	Name	Description
int?	messageTimeoutMs	The local message timeout.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithPartitioner(Partitioner?)

Sets the partitioner to be used to decide the target partition for a message: Confluent.Kafka.Partitioner.Random to randomly distribute the messages, Confluent.Kafka.Partitioner.Consistent to use the CRC32 hash of the message key (empty and null keys are mapped to a single partition), Confluent.Kafka.Partitioner.ConsistentRandom to use the CRC32 hash of the message key (but empty and null keys are randomly partitioned), Confluent.Kafka.Partitioner.Murmur2 to use a Java Producer compatible Murmur2 hash of the message key (null keys are mapped to a single partition), or Confluent.Kafka.Partitioner.Murmur2Random to use a Java Producer compatible Murmur2 hash of the message key (but null keys are randomly partitioned).
The default is Confluent.Kafka.Partitioner.ConsistentRandom, while Confluent.Kafka.Partitioner.Murmur2Random is functionally equivalent to the default partitioner in the Java Producer.

Declaration

public KafkaProducerConfigurationBuilder WithPartitioner(Partitioner? partitioner)

Parameters

Type	Name	Description
Partitioner?	partitioner	The partitioner to be used to decide the target partition for a message.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithQueueBufferingBackpressureThreshold(int?)

Sets the threshold of outstanding not yet transmitted broker requests needed to backpressure the producer's message accumulator. If the number of not yet transmitted requests equals or exceeds this number, produce request creation that would have otherwise been triggered (for example, in accordance with LingerMs) will be delayed. A lower number yields larger and more effective batches. A higher value can improve latency when using compression on slow machines.

Declaration

public KafkaProducerConfigurationBuilder WithQueueBufferingBackpressureThreshold(int? queueBufferingBackpressureThreshold)

Parameters

Type	Name	Description
int?	queueBufferingBackpressureThreshold	The threshold of outstanding not yet transmitted broker requests.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithQueueBufferingMaxKbytes(int?)

Sets the maximum total message size sum allowed on the producer queue. This queue is shared by all topics and partitions. This property has higher priority than QueueBufferingMaxMessages.

Declaration

public KafkaProducerConfigurationBuilder WithQueueBufferingMaxKbytes(int? queueBufferingMaxKbytes)

Parameters

Type	Name	Description
int?	queueBufferingMaxKbytes	The maximum total message size sum allowed on the producer queue.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithQueueBufferingMaxMessages(int?)

Sets the maximum number of messages allowed on the producer queue. This queue is shared by all topics and partitions.

Declaration

public KafkaProducerConfigurationBuilder WithQueueBufferingMaxMessages(int? queueBufferingMaxMessages)

Parameters

Type	Name	Description
int?	queueBufferingMaxMessages	The maximum number of messages allowed on the producer queue.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithRequestTimeoutMs(int?)

Sets the ack timeout of the producer request in milliseconds. This value is only enforced by the broker and relies on request.required.acks being != 0.

Declaration

public KafkaProducerConfigurationBuilder WithRequestTimeoutMs(int? requestTimeoutMs)

Parameters

Type	Name	Description
int?	requestTimeoutMs	The ack timeout of the producer request.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithSendOffsetToTransactionTimeout(TimeSpan)

Specifies the timeout for sending the offset to the transaction. The default is 10 seconds.

Declaration

public KafkaProducerConfigurationBuilder WithSendOffsetToTransactionTimeout(TimeSpan timeout)

Parameters

Type	Name	Description
TimeSpan	timeout	The timeout for sending the offset to the transaction.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithStickyPartitioningLingerMs(int?)

Sets the delay in milliseconds to wait to assign new sticky partitions for each topic. By default this is set to double the time of LingerMs. To disable sticky behavior, set it to 0. This behavior affects messages with the key null in all cases, and messages with key lengths of zero when the Confluent.Kafka.Partitioner.ConsistentRandom partitioner is in use. These messages would otherwise be assigned randomly. A higher value allows for more effective batching of these messages.

Declaration

public KafkaProducerConfigurationBuilder WithStickyPartitioningLingerMs(int? stickyPartitioningLingerMs)

Parameters

Type	Name	Description
int?	stickyPartitioningLingerMs	The delay in milliseconds to wait to assign new sticky partitions for each topic.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithTransactionAbortTimeout(TimeSpan)

Specifies the transaction abort operation timeout. The default is 30 seconds.

Declaration

public KafkaProducerConfigurationBuilder WithTransactionAbortTimeout(TimeSpan timeout)

Parameters

Type	Name	Description
TimeSpan	timeout	The transaction abort operation timeout.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithTransactionCommitTimeout(TimeSpan)

Specifies the transaction commit operation timeout. The default is 30 seconds.

Declaration

public KafkaProducerConfigurationBuilder WithTransactionCommitTimeout(TimeSpan timeout)

Parameters

Type	Name	Description
TimeSpan	timeout	The transaction commit operation timeout.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithTransactionTimeoutMs(int?)

Sets the maximum amount of time in milliseconds that the transaction coordinator will wait for a transaction status update from the producer before proactively aborting the ongoing transaction. If this value is larger than the transaction.max.timeout.ms setting in the broker, the init transaction call will fail with ERR_INVALID_TRANSACTION_TIMEOUT. The transaction timeout automatically adjusts MessageTimeoutMs and SocketTimeoutMs unless explicitly configured in which case they must not exceed the transaction timeout (SocketTimeoutMs must be at least 100ms lower than TransactionTimeoutMs).

Declaration

public KafkaProducerConfigurationBuilder WithTransactionTimeoutMs(int? transactionTimeoutMs)

Parameters

Type	Name	Description
int?	transactionTimeoutMs	The maximum amount of time that the transaction coordinator will wait for a transaction status update from the producer before proactively aborting the ongoing transaction.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.

WithTransactionsInitTimeout(TimeSpan)

Specifies the transactions init operation timeout. The default is 30 seconds.

Declaration

public KafkaProducerConfigurationBuilder WithTransactionsInitTimeout(TimeSpan timeout)

Parameters

Type	Name	Description
TimeSpan	timeout	The transactions init operation timeout.

Returns

Type	Description
KafkaProducerConfigurationBuilder	The KafkaProducerConfigurationBuilder so that additional calls can be chained.