Produce
Kafka produce operation.
Task version: 2.0.0
Required Frends version: 5.5+
Required .NET version: 6.0
Compatible Agents: Crossplatform
Source code: https://github.com/FrendsPlatform/Frends.Kafka
Task Parameters
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:
None: Compression codec to use for compressing message sets.Gzip: Compression codec to use for compressing message sets.Snappy: Compression codec to use for compressing message sets.Lz4: Compression codec to use for compressing message sets.Zstd: Compression codec to use for compressing message sets.
Default: 0
Example: CompressionTypes.None
SecurityProtocol : SecurityProtocols
Protocol used to communicate with brokers.
Possible values:
Plaintext: Protocol used to communicate with brokers.Ssl: Protocol used to communicate with brokers.SaslPlaintext: Protocol used to communicate with brokers.SaslSsl: Protocol used to communicate with brokers.
Default: 0
Example: SecurityProtocols.Plaintext
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
UseSasl : Boolean
Use SASL.
Default: False
Example: false
SaslMechanism : SaslMechanisms
SASL mechanism to use for authentication.
Possible values:
Gssapi: SASL mechanism to use for authentication.Plain: SASL mechanism to use for authentication.ScramSha256: SASL mechanism to use for authentication.ScramSha512: SASL mechanism to use for authentication.OAuthBearer: 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:
Default: 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.
Oidc: 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
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:
None: 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.
Https: 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
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: http://localhost:8081
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:
UserInfo: 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.
SaslInherit: 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!" } }
Acks : Ack
This field indicates the number of acknowledgements the leader broker must receive from ISR brokers before responding to the request.
Possible values:
None: This field indicates the number of acknowledgements the leader broker must receive from ISR brokers before responding to the request.Leader: This field indicates the number of acknowledgements the leader broker must receive from ISR brokers before responding to the request.All: 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:
Random: Partitioner.Consistent: Partitioner.ConsistentRandom: Partitioner.Murmur2: Partitioner.Murmur2Random: 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
Task Result
Success : Boolean
True if message was produced successfully.
Example: true
Data : Object
Produce result.
Example: PossiblyPersisted
Task Changelog
Changelog for Task Frends.Kafka.Produce.
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
Last updated
Was this helpful?