Consume
Kafka Consume 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
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
MessageCount : Int32
Amount of consumed messages before ending this task.
0 = unlimited, consume until timeout or task cancellation.
Default: -
Example: 10
Timeout : Int32
Consume operation timeout (value in ms).
0=unlimited. See other timeout options in Options-tab.
Default: 0
Example: 60000
Partition : Int32
Set Kafka partition.
Consume from all topic's partitions if set to -1.
Default: -
Example: 10
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.
Consumer: FetchRequests will use Options.FetchWaitMaxMs + Socket.SocketTimeoutMs.
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.
Not supported on Windows.
Default: -
Example: c:\temp
SaslKerberosMinTimeBeforeRelogin : Int32
Minimum time in milliseconds between key refresh attempts.
Disable automatic key refresh by setting this property to 0.
Not supported on Windows.
Default: -
Example: 60000
SaslKerberosPrincipal : String
This client's Kerberos principal name.
Not supported on Windows.
Default: -
Example: kafkaclient
SaslKerberosServiceName : String
Kerberos principal name that Kafka runs as, not including /hostname@REALM
Not supported on Windows.
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
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
EncodeMessageKey : Boolean
Try to decode consumed message key from byte[] to string.
Default: True
Example: true
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
AutoCommitIntervalMs : Int32
The frequency in milliseconds that the consumer offsets are committed (written) to offset storage.
(0 = disable). This setting is used by the high-level consumer.
Default: 5000
Example: 5000
AutoOffsetReset : AutoOffsetResets
Action to take when there is no initial offset in offset store or the desired offset is out of range.
Possible values:
Latest: Action to take when there is no initial offset in offset store or the desired offset is out of range.Earliest: Action to take when there is no initial offset in offset store or the desired offset is out of range.Error: Action to take when there is no initial offset in offset store or the desired offset is out of range.
Default: 0
Example: AutoOffsetResets.
EnableAutoOffsetStore : Boolean
Automatically store offset of last message provided to application.
The offset store is an in-memory store of the next offset to (auto-)commit for each partition.
Default: True
Example: true
BrokerAddressFamily : BrokerAddressFamilys
Allowed broker IP address families.
Possible values:
Any: Allowed broker IP address families.V4: Allowed broker IP address families.V6: Allowed broker IP address families.
Default: 0
Example: BrokerAddressFamilys.Any
ConnectionsMaxIdleMs : Int32
Close broker connections after the specified time of inactivity.
Disable with 0.
If this property is left at its default value some heuristics are performed to determine a suitable default value, this is currently limited to identifying brokers on Azure.
Default: 0
Example: 0
CheckCrcs : Boolean
Verify CRC32 of consumed messages, ensuring no on-the-wire or on-disk corruption to the messages occurred.
Default: False
Example: false
EnableAutoCommit : Boolean
Automatically and periodically commit offsets in the background.
Note: setting this to false does not prevent the consumer from fetching previously committed start offsets.
Default: True
Example: true
FetchErrorBackoffMs : Int32
How long to postpone the next fetch request for a topic+partition in case of a fetch error.
Default: 500
Example: 500
FetchMaxBytes : Int32
Maximum amount of data the broker shall return for a Fetch request.
Messages are fetched in batches by the consumer and if the first message batch in the first non-empty partition of the Fetch request is larger than this value, then the message batch will still be returned to ensure the consumer can make progress.
The maximum message batch size accepted by the broker is defined via message.max.bytes (broker config) or max.message.bytes (broker topic config).
FetchMaxBytes is automatically adjusted upwards to be at least Options.MessageMaxBytes
Default: 52428800
Example: 52428800
FetchMinBytes : Int32
Minimum number of bytes the broker responds with.
If option.FetchWaitMaxMs expires the accumulated data will be sent to the client regardless of this setting.
Default: 1
Example: 1
FetchWaitMaxMs : Int32
Maximum time the broker may wait to fill the Fetch response with Options.FetchMinBytes of messages.
Default: 500
Example: 500
GroupId : String
Client group id string.
All clients sharing the same Options.GroupId belong to the same group.
Default: -
Example: csharp-group-1
GroupInstanceId : String
Enable static group membership.
Static group members are able to leave and rejoin a group within the configured Options.SessionTimeoutMs without prompting a group rebalance.
This should be used in combination with a larger Options.SessionTimeoutMs to avoid group rebalances caused by transient unavailability (e.g. process restarts).
Requires broker version >= 2.3.0.
Default: -
Example: csharp-group-1
HeartbeatIntervalMs : Int32
Group session keepalive heartbeat interval.
Default: 3000
Example: 3000
IsolationLevel : IsolationLevels
Controls how to read messages written transactionally.
Possible values:
ReadUncommitted: Controls how to read messages written transactionally.ReadCommitted: Controls how to read messages written transactionally.
Default: 1
Example: IsolationLevels.ReadCommitted
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
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
MaxPollIntervalMs : Int32
Maximum allowed time between calls to consume messages for high-level consumers.
If this interval is exceeded the consumer is considered failed and the group will rebalance in order to reassign the partitions to another consumer group member.
Warning: Offset commits may be not possible at this point.
Options.MaxPollIntervalMs must be >= Options.SessionTimeoutMs.
The interval is checked two times per second.
Default: 300000
Example: 300000
QueuedMaxMessagesKbytes : Int32
Maximum number of kilobytes of queued pre-fetched messages in the local consumer queue.
If using the high-level consumer this setting applies to the single consumer queue, regardless of the number of partitions.
When using the legacy simple consumer or when separate partition queues are used this setting applies per partition.
This value may be overshot by Options.FetchMaxBytes.
This property has higher priority than Options.QueuedMinMessages.
Default: 65536
Example: 65536
QueuedMinMessages : Int32
Minimum number of messages per topic+partition librdkafka tries to maintain in the local consumer queue.
Default: 100000
Example: 100000
ReconnectBackoffMaxMs : Int32
The maximum time to wait before reconnecting to a broker after the connection has been closed.
Default: 10000
Example: 10000
ReconnectBackoffMs : Int32
The initial time to wait before reconnecting to a broker after the connection has been closed.
The time is increased exponentially until Options.ReconnectBackoffMaxMs is reached.
-25% to +50% jitter is applied to each reconnect backoff.
A value of 0 disables the backoff and reconnects immediately.
Default: 100
Example: 100
SessionTimeoutMs : Int32
Client group session and failure detection timeout.
The consumer sends periodic heartbeats (Options.HeartbeatIntervalMs) to indicate its liveness to the broker.
If no hearts are received by the broker for a group member within the session timeout, the broker will remove the consumer from the group and trigger a rebalance.
Options.MaxPollIntervalMs must be >= Options.SessionTimeoutMs.
Default: 45000
Example: 45000
Debug : String
A comma-separated list of debug contexts to enable.
Detailed Consumer debugging: consumer,cgrp,topic,fetch.
Default: -
Example: broker
Task Result
Success : Boolean
True if messages have been consumed without errors.
Example: true
Data : List<Message>
Result data.
Example: Object { key, value }
Task Changelog
Changelog for Task Frends.Kafka.Consume.
Changelog
[2.0.0] - 2024-05-15
Added
Support for Confluent Schema Registry based Avro.
New parameter Options.Debug.
Changed
Confluent.Kafka updated from version 1.9.3 to 2.4.
Input.Partition change: Consume from all topic's partitions if set to -1.
New parameter: Options.EncodeMessageKey to choose whether this Task will try to encode consumed key to string from byte[]
Message class change: string Key, string Value replaced by dynamic Key, dynamic Value.
Result.Messages renamed to Result.Data.
Removed optional parameters: ApiVersionRequest, ApiVersionFallbackMs, ApiVersionRequestTimeoutMs, AllowAutoCreateTopics,
[1.1.0] - 2023-11-27
Added
Added a partition as input parameter to the task.
[1.0.1] - 2023-04-17
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?