# Produce Required Frends version: 5.5+ Required .NET version: 6.0 Compatible Agents: Crossplatform Source code: {% tabs %} {% tab title="Parameter: Input" %} | Name | Description | | -------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Host : `String` |

Initial list of brokers as a CSV list of broker host or host:port.


Default: -
Example: localhost:1234

| | Topic : `String` |

Topic.


Default: -
Example: ExampleTopic

| | Key : `String` |

Message key value.

Can be empty.


Default: -
Example: examplekey

| | Message : `String` |

Message.

Can be empty.


Default: -
Example: Example message.

| | Partition : `Int32` |

The partition value.

If set to -1. The partition the message is sent to is determined by the partitioner defined using the Options.Partitioner configuration property.


Default: 0
Example: 0

| | CompressionType : `CompressionTypes` |

Compression codec to use for compressing message sets.

Possible values:

Compression codec to use for compressing message sets.

Compression codec to use for compressing message sets.

Compression codec to use for compressing message sets.

Compression codec to use for compressing message sets.

Compression codec to use for compressing message sets.


Default: 0
Example: CompressionTypes.None

| | SecurityProtocol : `SecurityProtocols` |

Protocol used to communicate with brokers.

Possible values:

Protocol used to communicate with brokers.

Protocol used to communicate with brokers.

Protocol used to communicate with brokers.

Protocol used to communicate with brokers.


Default: 0
Example: SecurityProtocols.Plaintext

| | {% endtab %} | | {% tab title="Parameter: Socket" %} | Name | Description | | ---------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | SocketKeepaliveEnable : `Boolean` |

Enable TCP keep-alives (SO\_KEEPALIVE) on broker sockets.


Default: False
Example: false

| | SocketNagleDisable : `Boolean` |

Disable the Nagle algorithm (TCP\_NODELAY) on broker sockets.


Default: False
Example: false

| | SocketTimeoutMs : `Int32` |

Default timeout for network requests.

ProduceRequests will use the lesser value of Socket.SocketTimeoutMs and remaining Options.MessageTimeoutMs for the first message in the batch.


Default: 60000
Example: 60000

| | SocketConnectionSetupTimeoutMs : `Int32` |

Maximum time allowed for broker connection setup (TCP connection setup as well SSL and SASL handshake).

If the connection to the broker is not fully functional after this the connection will be closed and retried.


Default: 30000
Example: 30000

| | SocketMaxFails : `Int32` |

Disconnect from broker when this number of Produce failures (e.g., timed out requests) is reached.

Disable with 0.

WARNING: It is highly recommended to leave this setting at its default value of 1 to avoid the client and broker to become desynchronized in case of request timeouts.

NOTE: The connection is automatically re-established.


Default: 1
Example: 1

| | SocketReceiveBufferBytes : `Int32` |

Broker socket receive buffer size.

0 = System default.


Default: 0
Example: 0

| | {% endtab %} | | {% tab title="Parameter: Sasl" %} | Name | Description | | ------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | UseSasl : `Boolean` |

Use SASL.


Default: False
Example: false

| | SaslMechanism : `SaslMechanisms` |

SASL mechanism to use for authentication.

Possible values:

SASL mechanism to use for authentication.

SASL mechanism to use for authentication.

SASL mechanism to use for authentication.

SASL mechanism to use for authentication.

SASL mechanism to use for authentication.


Default: 1
Example: SaslMechanisms.Plain

| | SaslUsername : `String` |

SASL username for use with the PLAIN, ScramSha256 or ScramSha512 mechanism.


Default: -
Example: ExampleUser

| | đź—ťSaslPassword : `String` |

SASL password for use with the PLAIN, ScramSha256 or ScramSha512 mechanism.


Default: -
Example: ExamplePassword

| | SaslOauthbearerMethod : `SaslOauthbearerMethods` |

Set to "default" or "oidc" to control which login method to be used.

If set to "oidc", the following properties must also be be specified:

Sasl.SaslOauthbearerClientId,

Sasl.SaslOauthbearerClientSecret,

Sasl.SaslOauthbearerTokenEndpointUrl.

Possible values:

Set to "default" or "oidc" to control which login method to be used.

If set to "oidc", the following properties must also be be specified:

Sasl.SaslOauthbearerClientId,

Sasl.SaslOauthbearerClientSecret,

Sasl.SaslOauthbearerTokenEndpointUrl.

Set to "default" or "oidc" to control which login method to be used.

If set to "oidc", the following properties must also be be specified:

Sasl.SaslOauthbearerClientId,

Sasl.SaslOauthbearerClientSecret,

Sasl.SaslOauthbearerTokenEndpointUrl.


Default: 0
Example: SaslOauthbearerMethod.Default

| | SaslOauthbearerClientId : `String` |

Public identifier for the application.

Must be unique across all clients that the authorization server handles.


Default: -
Example: ExampleClient

| | đź—ťSaslOauthbearerClientSecret : `String` |

Client secret only known to the application and the authorization server.


Default: -
Example: ExampleSecret

| | SaslOauthbearerTokenEndpointUrl : `String` |

OAuth/OIDC issuer token endpoint HTTP(S) URI used to retrieve token.


Default: -
Example: ExampleURL

| | SaslOauthbearerExtensions : `String` |

Allow additional information to be provided to the broker.

Comma-separated list of key=value pairs. E.g., "supportFeatureX=true,organizationId=sales-emea".


Default: -
Example: supportFeatureX=true,organizationId=sales-emea

| | SaslOauthbearerScope : `String` |

Client use this to specify the scope of the access request to the broker.


Default: -
Example: ExampleScope

| | SaslOauthbearerConfig : `String` |

SASL/OAUTHBEARER configuration.


Default: -
Example: principal=admin extension\_traceId=123

| | SaslKerberosKeytab : `String` |

Path to Kerberos keytab file.

Only supported on Linux OS.


Default: -
Example: /path/to/your/file

| | SaslKerberosMinTimeBeforeRelogin : `Int32` |

Minimum time in milliseconds between key refresh attempts.

Disable automatic key refresh by setting this property to 0.

Only supported on Linux OS.


Default: -
Example: 60000

| | SaslKerberosPrincipal : `String` |

This client's Kerberos principal name.

Only supported on Linux OS.


Default: -
Example: kafkaclient

| | SaslKerberosServiceName : `String` |

Kerberos principal name that Kafka runs as, not including /hostname\@REALM

Only supported on Linux OS.


Default: -
Example: kafka

| | {% endtab %} | | {% tab title="Parameter: Ssl" %} | Name | Description | | -------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | UseSsl : `Boolean` |

Use SSL.


Default: False
Example: false

| | SslEndpointIdentificationAlgorithm : `SslEndpointIdentificationAlgorithms` |

Endpoint identification algorithm to validate broker hostname using broker certificate.

https - Server (broker) hostname verification as specified in RFC2818.

none - No endpoint verification.

OpenSSL >= 1.0.2 required.

Possible values:

Endpoint identification algorithm to validate broker hostname using broker certificate.

https - Server (broker) hostname verification as specified in RFC2818.

none - No endpoint verification.

OpenSSL >= 1.0.2 required.

Endpoint identification algorithm to validate broker hostname using broker certificate.

https - Server (broker) hostname verification as specified in RFC2818.

none - No endpoint verification.

OpenSSL >= 1.0.2 required.


Default: 0
Example: SslEndpointIdentificationAlgorithm.None

| | EnableSslCertificateVerification : `Boolean` |

Enable OpenSSL's builtin broker (server) certificate verification.


Default: True
Example: true

| | SslCertificateLocation : `String` |

Path to client's public key (PEM) used for authentication.


Default: -
Example: /path/to/your/file

| | SslCaLocation : `String` |

File or directory path to CA certificate(s) for verifying the broker's key.


Default: -
Example: Root

| | SslKeyLocation : `String` |

Path to client's private key (PEM) used for authentication.


Default: -
Example: /path/to/your/file

| | SslKeystoreLocation : `String` |

Path to client's keystore (PKCS#12) used for authentication.


Default: -
Example: /path/to/your/file

| | SslEngineLocation : `String` |

Path to OpenSSL engine library.

OpenSSL >= 1.1.0 required.


Default: -
Example: /path/to/your/file

| | SslCrlLocation : `String` |

Path to CRL for verifying broker's certificate validity.


Default: -
Example: /path/to/your/file

| | đź—ťSslCertificatePem : `String` |

Client's public key string (PEM format) used for authentication.


Default: -
Example: —–BEGIN PRIVATE KEY—–MIIES42Cg6zn—–END PRIVATE KEY—–

| | đź—ťSslCaPem : `String` |

A certificate string (PEM format) for verifying the broker's key.


Default: -
Example: —–BEGIN PRIVATE KEY—–MIIES42Cg6zn—–END PRIVATE KEY—–

| | đź—ťSslKeyPem : `String` |

Client's private key string (PEM format) used for authentication.


Default: -
Example: —–BEGIN PRIVATE KEY—–MIIES42Cg6zn—–END PRIVATE KEY—–

| | SslCaCertificateStores : `String` |

Comma-separated list of Windows Certificate stores to load CA certificates from.

Certificates will be loaded in the same order as stores are specified.

If no certificates can be loaded from any of the specified stores an error is logged and the OpenSSL library's default CA location is used instead.


Default: Root
Example: Root

| | đź—ťSslKeystorePassword : `String` |

Client's keystore (PKCS#12) password.


Default: -
Example: ExamplePassword

| | đź—ťSslKeyPassword : `String` |

Private key passphrase for use with Ssl.SslKeyLocation)


Default: -
Example: ExamplePassword

| | SslCipherSuites : `String` |

A cipher suite is a named combination of authentication, encryption, MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol.


Default: -
Example: foo

| | SslCurvesList : `String` |

The supported-curves extension in the TLS ClientHello message specifies the curves (standard/named, or 'explicit' GF(2^k) or GF(p)) the client is willing to have the server use.


Default: -
Example: 1

| | SslSigalgsList : `String` |

The client uses the TLS ClientHello signature\_algorithms extension to indicate to the server which signature/hash algorithm pairs may be used in digital signatures.


Default: -
Example: 1

| | {% endtab %} | | {% tab title="Parameter: SchemaRegistry" %} | Name | Description | | ----------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | UseSchemaRegistry : `Boolean` |

Use Avro schema registry.


Default: False
Example: false

| | SchemaRegistryUrl : `String` |

A comma-separated list of URLs for schema registry instances that are used to register or lookup schemas.


Default: -
Example:

| | BasicAuthCredentialsSource : `AuthCredentialsSources` |

Authentication credentials source.

UserInfo = Credentials are specified via the BasicAuthUserInfo property in the form username:password. If BasicAuthUserInfo is not set, authentication is disabled.

SaslInherit = Credentials are specified via the Sasl.SaslUsername and Sasl.SaslPassword parameters.

Possible values:

Authentication credentials source.

UserInfo = Credentials are specified via the BasicAuthUserInfo property in the form username:password. If BasicAuthUserInfo is not set, authentication is disabled.

SaslInherit = Credentials are specified via the Sasl.SaslUsername and Sasl.SaslPassword parameters.

Authentication credentials source.

UserInfo = Credentials are specified via the BasicAuthUserInfo property in the form username:password. If BasicAuthUserInfo is not set, authentication is disabled.

SaslInherit = Credentials are specified via the Sasl.SaslUsername and Sasl.SaslPassword parameters.


Default: 0
Example: AuthCredentialsSource.UserInfo

| | đź—ťBasicAuthUserInfo : `String` |

Basic auth credentials in the form {username}:{password}.


Default: -
Example: foo:bar

| | EnableSslCertificateVerification : `Boolean` |

Enable SSL verification.

Disabling SSL verification is insecure and should only be done for reasons of convenience in test/dev environments.


Default: True
Example: true

| | MaxCachedSchemas : `Int32` |

Specifies the maximum number of schemas CachedSchemaRegistryClient should cache locally.


Default: 1000
Example: 1000

| | RequestTimeoutMs : `Int32` |

Specifies the timeout for requests to Confluent Schema Registry.


Default: 30000
Example: 30000

| | SslCaLocation : `String` |

File path to CA certificate(s) for verifying the Schema Registry's key.

System CA certs will be used if not specified.


Default: -
Example: /path/to/your/file

| | SslKeystoreLocation : `String` |

SSL keystore (PKCS#12) location.


Default: -
Example: /path/to/your/file

| | SslKeystorePassword : `String` |

SSL keystore (PKCS#12) password.


Default: -
Example: foo

| | SchemaJson : `String` |

Schema JSON.


Default: -
Example: { "fields": \[ { "name": "intField", "type": "int" }, { "name": "longField", "type": "long" }, { "name": "floatField", "type": "float" }, { "name": "doubleField", "type": "double" }, { "name": "booleanField", "type": "boolean" }, { "name": "stringField", "type": "string" }, { "name": "nullField", "type": "null" }, { "name": "bytesField", "type": "bytes" }, { "name": "enumField", "type": { "name": "Colors", "symbols": \[ "RED", "GREEN", "BLUE" ], "type": "enum" } }, { "name": "arrayField", "type": { "items": "string", "type": "array" } }, { "name": "mapField", "type": { "type": "map", "values": "int" } }, { "name": "fixedField", "type": { "name": "FourBytes", "size": 4, "type": "fixed" } }, { "name": "unionField", "type": \[ "null", "string" ] }, { "name": "recordField", "type": { "fields": \[ { "name": "nestedField", "type": "string" } ], "name": "NestedRecord", "type": "record" } } ], "name": "sampleRecord", "namespace": "com.mycorp.mynamespace", "type": "record" }

| | Records : `String` |

Field values as an JSON array.


Default: -
Example: { "intField": 123, "longField": 1234567890, "floatField": 1.23, "doubleField": 1.23456789, "booleanField": true, "stringField": "Hello, World!", "nullField": null, "bytesField": "dGVzdDE=", "enumField": "RED", "arrayField": \["item1", "item2", "item3"], "mapField": { "key1": 1, "key2": 2, "key3": 3 }, "fixedField": "YWJjZA==", "unionField": "Hello, Union!", "recordField": { "nestedField": "Hello, Nested!" } }

| | {% endtab %} | | {% tab title="Parameter: Options" %} | Name | Description | | ----------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Acks : `Ack` |

This field indicates the number of acknowledgements the leader broker must receive from ISR brokers before responding to the request.

Possible values:

This field indicates the number of acknowledgements the leader broker must receive from ISR brokers before responding to the request.

This field indicates the number of acknowledgements the leader broker must receive from ISR brokers before responding to the request.

This field indicates the number of acknowledgements the leader broker must receive from ISR brokers before responding to the request.


Default: 0
Example: Ack.None

| | EnableIdempotence : `Boolean` |

When set to true, 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:

Options.MaxInFlight=5 (must be less than or equal to 5),

Options.MessageProduceMaxRetries=2147483647 (must be greater than 0),

Options.Ack.All,

Producer instantation will fail if user-supplied configuration is incompatible.


Default: False
Example: false

| | LingerMs : `Int32` |

Delay in milliseconds to wait for messages in the producer queue to accumulate before constructing message batches (MessageSets) 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.


Default: 5
Example: 5

| | MaxInFlight : `Int32` |

Maximum number of in-flight requests per broker connection.

This is a generic property applied to all broker communication, however it is primarily relevant to produce requests.

In particular, note that other mechanisms limit the number of outstanding consumer fetch request per broker to one.


Default: 1000000
Example: 1000000

| | MessageTimeoutMs : `Int32` |

Local message timeout. 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 librdkafka may use to deliver a message (including retries).

Delivery error occurs when either the retry count or the message timeout are exceeded.

The message timeout is automatically adjusted to TransactionTimeoutMs if Options.TransactionalId is configured.


Default: 300000
Example: 300000

| | MessageMaxBytes : `Int32` |

Maximum Kafka protocol request message size.

Due to differing framing overhead between protocol versions the producer is unable to reliably enforce a strict max message limit at produce time and may exceed the maximum size by one message in protocol ProduceRequests, the broker will enforce the the topic's Options.MessageMaxBytes limit (see Apache Kafka documentation)


Default: 1000000
Example: 1000000

| | MessageSendMaxRetries : `Int32` |

How many times to retry sending a failing Message.

Retrying may cause reordering unless Options.EnableIdempotence is set to true.


Default: 2147483647
Example: -

| | Partitioner : `Partitioners` |

Partitioner.

Possible values:

Partitioner.

Partitioner.

Partitioner.

Partitioner.

Partitioner.


Default: 2
Example: Partitioners.ConsistentRandom

| | QueueBufferingMaxKbytes : `Int32` |

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 Options.QueueBufferingMaxMessages.


Default: 1048576
Example: 1048576

| | QueueBufferingMaxMessages : `Int32` |

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


Default: 100000
Example: 100000

| | TransactionalId : `String` |

Enables the transactional producer.

Used to identify the same transactional producer instance across process restarts.

It allows the producer to guarantee that transactions corresponding to earlier instances of the same producer have been finalized prior to starting any new transactions, and that any zombie instances are fenced off.

If no TransactionalId is provided, then the producer is limited to idempotent delivery (if Options.EnableIdempotence is set). Requires broker version >= 0.11.0.


Default: -
Example: 1

| | TransactionTimeoutMs : `Int32` |

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 call will fail with Timeout error.

The transaction timeout automatically adjusts Options.MessageTimeoutMs and Socket.SocketTimeoutMs, unless explicitly configured in which case they must not exceed the transaction timeout (Socket.SocketTimeoutMs must be at least 100ms lower than Options.TransactionTimeoutMs).

This is also the default timeout value if no timeout (-1) is supplied to the transactional API methods.


Default: 60000
Example: 60000

| | Debug : `String` |

A comma-separated list of debug contexts to enable.

Detailed Producer debugging: broker,topic,msg.


Default: -
Example: broker

| | {% endtab %} | | {% tab title="Result: Result" %} | Name | Description | | ------------------- | ------------------------------------------------------------------------------- | | Success : `Boolean` |

True if message was produced successfully.
Example: true

| | Data : `Object` |

Produce result.
Example: PossiblyPersisted

| | {% endtab %} | | {% tab title="Changelog" %} ## Changelog ### \[2.0.0] - 2024-05-14 #### Added * Support for Confluent Schema Registry based Avro. * New parameter Options.Debug. #### Changed * Result.Status and Result.Timestamp have been replaced by Result.Data. * Changed how partitions are handled when producing message. See Input.Partition description. * All but Kerberos SASL settings can be used on Windows platform. * Confluent.Kafka updated from version 1.9.3 to 2.4. * Parameter removed: Options.ApiVersionRequest. ### \[1.2.0] - 2023-10-11 #### Changed * Input.Message will no longer be serialized into JSON text before sending. ### \[1.1.0] - 2023-09-01 #### Added * Input.Partition, set the partition. * Input.Key, set message key. ### \[1.0.1] - 2023-04-04 #### Fixed * Changed Task to set ssl.SslCaCertificateStores only if it's set as parameter. ### \[1.0.0] - 2022-10-19 #### Added * Initial implementation {% endtab %} {% endtabs %}