Class KafkaProducerConfigurationBuilder
- Namespace
- Silverback.Messaging.Configuration.Kafka
- Assembly
- Silverback.Integration.Kafka.dll
Builds the KafkaProducerConfiguration.
public class KafkaProducerConfigurationBuilder : KafkaClientConfigurationBuilder<KafkaProducerConfiguration, ProducerConfig, KafkaProducerConfigurationBuilder>- Inheritance
-
KafkaClientConfigurationBuilder<KafkaProducerConfiguration, ProducerConfig, KafkaProducerConfigurationBuilder>KafkaProducerConfigurationBuilder
- Inherited Members
Constructors
KafkaProducerConfigurationBuilder(IServiceProvider)
Initializes a new instance of the KafkaProducerConfigurationBuilder class.
public KafkaProducerConfigurationBuilder(IServiceProvider serviceProvider)Parameters
serviceProviderIServiceProviderThe IServiceProvider.
Properties
This
Gets this instance.
protected override KafkaProducerConfigurationBuilder This { get; }Property Value
Remarks
This is necessary to work around casting in the base classes.
Methods
Build()
Builds the KafkaProducerConfiguration instance.
public KafkaProducerConfiguration Build()Returns
BuildCore()
Builds the configuration.
protected override KafkaProducerConfiguration BuildCore()Returns
- KafkaProducerConfiguration
The configuration.
DisableDeliveryReports()
Disable notification of delivery reports. This will enable "fire and forget" semantics and give a small boost in performance.
public KafkaProducerConfigurationBuilder DisableDeliveryReports()Returns
- 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.
public KafkaProducerConfigurationBuilder DisableDisposeOnException()Returns
- KafkaProducerConfigurationBuilder
The KafkaProducerConfigurationBuilder so that additional calls can be chained.
DisableGaplessGuarantee()
Disables the gapless guarantee.
public KafkaProducerConfigurationBuilder DisableGaplessGuarantee()Returns
- 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.
public KafkaProducerConfigurationBuilder DisableIdempotence()Returns
- KafkaProducerConfigurationBuilder
The KafkaProducerConfigurationBuilder so that additional calls can be chained.
DisableTransactions()
Disables the Kafka transactions. The producer is limited to idempotent delivery (see EnableIdempotence()).
public KafkaProducerConfigurationBuilder DisableTransactions()Returns
- 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.
public KafkaProducerConfigurationBuilder DisposeOnException()Returns
- KafkaProducerConfigurationBuilder
The KafkaProducerConfigurationBuilder so that additional calls can be chained.
EnableDeliveryReports()
Enable notification of delivery reports.
public KafkaProducerConfigurationBuilder EnableDeliveryReports()Returns
- 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.
public KafkaProducerConfigurationBuilder EnableGaplessGuarantee()Returns
- 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.
public KafkaProducerConfigurationBuilder EnableIdempotence()Returns
- 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.
public KafkaProducerConfigurationBuilder EnableTransactions(string transactionalId)Parameters
transactionalIdstringThe identifier to be used to identify the same transactional producer instance across process restarts.
Returns
- 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.
public KafkaProducerConfigurationBuilder IgnoreIfNotAcknowledged()Returns
- 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.).
public KafkaProducerConfigurationBuilder Produce(Action<KafkaProducerEndpointConfigurationBuilder<object>> configurationBuilderAction)Parameters
configurationBuilderActionAction<KafkaProducerEndpointConfigurationBuilder<object>>An Action that takes the KafkaProducerConfigurationBuilder and configures it.
Returns
- 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.).
public KafkaProducerConfigurationBuilder Produce(string? name, Action<KafkaProducerEndpointConfigurationBuilder<object>> configurationBuilderAction)Parameters
namestringThe 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.
configurationBuilderActionAction<KafkaProducerEndpointConfigurationBuilder<object>>An Action that takes the KafkaProducerConfigurationBuilder and configures it.
Returns
- 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.).
public KafkaProducerConfigurationBuilder Produce<TMessage>(Action<KafkaProducerEndpointConfigurationBuilder<TMessage>> configurationBuilderAction) where TMessage : classParameters
configurationBuilderActionAction<KafkaProducerEndpointConfigurationBuilder<TMessage>>An Action that takes the KafkaProducerConfigurationBuilder and configures it.
Returns
- KafkaProducerConfigurationBuilder
The KafkaProducerConfigurationBuilder so that additional calls can be chained.
Type Parameters
TMessageThe 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.).
public KafkaProducerConfigurationBuilder Produce<TMessage>(string? name, Action<KafkaProducerEndpointConfigurationBuilder<TMessage>> configurationBuilderAction) where TMessage : classParameters
namestringThe 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.
configurationBuilderActionAction<KafkaProducerEndpointConfigurationBuilder<TMessage>>An Action that takes the KafkaProducerConfigurationBuilder and configures it.
Returns
- KafkaProducerConfigurationBuilder
The KafkaProducerConfigurationBuilder so that additional calls can be chained.
Type Parameters
TMessageThe 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.
public KafkaProducerConfigurationBuilder ThrowIfNotAcknowledged()Returns
- 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.
public KafkaProducerConfigurationBuilder WithBatchNumMessages(int? batchNumMessages)Parameters
batchNumMessagesint?The maximum number of messages batched in one message set.
Returns
- 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.
public KafkaProducerConfigurationBuilder WithBatchSize(int? batchSize)Parameters
batchSizeint?The maximum size of all messages batched in one message set.
Returns
- 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.
public KafkaProducerConfigurationBuilder WithCompressionLevel(int? compressionLevel)Parameters
compressionLevelint?The compression level parameter for the algorithm selected by configuration property Confluent.Kafka.CompressionType.
Returns
- 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.
public KafkaProducerConfigurationBuilder WithCompressionType(CompressionType? compressionType)Parameters
compressionTypeCompressionType?The compression codec to be used to compress message sets.
Returns
- KafkaProducerConfigurationBuilder
The KafkaProducerConfigurationBuilder so that additional calls can be chained.
WithFlushTimeout(TimeSpan)
Specifies the flush operation timeout. The default is 30 seconds.
public KafkaProducerConfigurationBuilder WithFlushTimeout(TimeSpan timeout)Parameters
timeoutTimeSpanThe flush operation timeout.
Returns
- 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.
public KafkaProducerConfigurationBuilder WithLingerMs(double? lingerMs)Parameters
lingerMsdouble?The delay to wait for messages in the producer queue to accumulate before constructing message batches.
Returns
- 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.
public KafkaProducerConfigurationBuilder WithMessageSendMaxRetries(int? messageSendMaxRetries)Parameters
messageSendMaxRetriesint?How many times to retry sending a failing message.
Returns
- 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.
public KafkaProducerConfigurationBuilder WithMessageTimeoutMs(int? messageTimeoutMs)Parameters
messageTimeoutMsint?The local message timeout.
Returns
- 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.
public KafkaProducerConfigurationBuilder WithPartitioner(Partitioner? partitioner)Parameters
partitionerPartitioner?The partitioner to be used to decide the target partition for a message.
Returns
- 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.
public KafkaProducerConfigurationBuilder WithQueueBufferingBackpressureThreshold(int? queueBufferingBackpressureThreshold)Parameters
queueBufferingBackpressureThresholdint?The threshold of outstanding not yet transmitted broker requests.
Returns
- 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.
public KafkaProducerConfigurationBuilder WithQueueBufferingMaxKbytes(int? queueBufferingMaxKbytes)Parameters
queueBufferingMaxKbytesint?The maximum total message size sum allowed on the producer queue.
Returns
- 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.
public KafkaProducerConfigurationBuilder WithQueueBufferingMaxMessages(int? queueBufferingMaxMessages)Parameters
queueBufferingMaxMessagesint?The maximum number of messages allowed on the producer queue.
Returns
- 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.
public KafkaProducerConfigurationBuilder WithRequestTimeoutMs(int? requestTimeoutMs)Parameters
requestTimeoutMsint?The ack timeout of the producer request.
Returns
- 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.
public KafkaProducerConfigurationBuilder WithSendOffsetToTransactionTimeout(TimeSpan timeout)Parameters
timeoutTimeSpanThe timeout for sending the offset to the transaction.
Returns
- 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.
public KafkaProducerConfigurationBuilder WithStickyPartitioningLingerMs(int? stickyPartitioningLingerMs)Parameters
stickyPartitioningLingerMsint?The delay in milliseconds to wait to assign new sticky partitions for each topic.
Returns
- KafkaProducerConfigurationBuilder
The KafkaProducerConfigurationBuilder so that additional calls can be chained.
WithTransactionAbortTimeout(TimeSpan)
Specifies the transaction abort operation timeout. The default is 30 seconds.
public KafkaProducerConfigurationBuilder WithTransactionAbortTimeout(TimeSpan timeout)Parameters
timeoutTimeSpanThe transaction abort operation timeout.
Returns
- KafkaProducerConfigurationBuilder
The KafkaProducerConfigurationBuilder so that additional calls can be chained.
WithTransactionCommitTimeout(TimeSpan)
Specifies the transaction commit operation timeout. The default is 30 seconds.
public KafkaProducerConfigurationBuilder WithTransactionCommitTimeout(TimeSpan timeout)Parameters
timeoutTimeSpanThe transaction commit operation timeout.
Returns
- 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).
public KafkaProducerConfigurationBuilder WithTransactionTimeoutMs(int? transactionTimeoutMs)Parameters
transactionTimeoutMsint?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
- KafkaProducerConfigurationBuilder
The KafkaProducerConfigurationBuilder so that additional calls can be chained.
WithTransactionsInitTimeout(TimeSpan)
Specifies the transactions init operation timeout. The default is 30 seconds.
public KafkaProducerConfigurationBuilder WithTransactionsInitTimeout(TimeSpan timeout)Parameters
timeoutTimeSpanThe transactions init operation timeout.
Returns
- KafkaProducerConfigurationBuilder
The KafkaProducerConfigurationBuilder so that additional calls can be chained.