# Consume 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

| | 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

| | 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

| | {% 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.

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

| | {% 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.

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

| | {% 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

| | 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

| | {% endtab %} | | {% tab title="Parameter: Options" %} | Name | Description | | -------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | 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:

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

| | 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:

Action to take when there is no initial offset in offset store or the desired offset is out of range.

Action to take when there is no initial offset in offset store or the desired offset is out of range.

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:

Allowed broker IP address families.

Allowed broker IP address families.

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:

Controls how to read messages written transactionally.

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

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

True if messages have been consumed without errors.
Example: true

| | Data : `List` |

Result data.
Example: Object { key, value }

| | {% endtab %} | | {% tab title="Changelog" %} ## 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 {% endtab %} {% endtabs %}