Kafka
This connector allows you to send and receive messages through Kafka.
Asset Setup or Prerequisites
The connector asset requires the following inputs for connecting to kafka:
- username
- password
- ssl_certfile
- bootstrap_servers(required)
Capabilities
This connector provides the following capabilities:
- Get Data
- Send Data
Retrieve Data Task
Get Data tasks will track the offset for each partition they pull from. This prevents any messages from being pulled twice.
Notes
Check here To access kafka-python API Documentation for the Connector.
Configurations
Kafka API Authentication
Authenticates using Kafka API Authentication
Configuration Parameters
Parameter | Description | Type | Required |
|---|---|---|---|
bootstrap_servers | Comma separated brokers list. | string | Required |
username | SASL Username for Kafka. | string | Optional |
password | SASL Password for Kafka. | string | Optional |
ssl_certfile | Base64 encoded certificate in pem format to authenticate. | string | Optional |
verify_ssl | Verify SSL certificate | boolean | Optional |
http_proxy | A proxy to route requests through. | string | Optional |
Actions
Get Data
Gets data from a Kafka Topic.
Input
Argument Name | Type | Required | Description |
|---|---|---|---|
topic | string | Required | Kafka Topic. |
partitions | string | Optional | Comma separated partitions. |
partition_offsets | object | Optional | Partition Offsets which is an object where keys are partition ids and values are offsets. |
limit | number | Optional | Limit the number of messages to be retrieved. Defaults to 100. |
return_single_record | boolean | Optional | If true, return all the results in a single record in messages in result object. If false, create a record for each result with the message and returns data in messages in result object. |
client_id | string | Optional | A name for this client. |
group_id | string | Optional | The name of the consumer group to join for dynamic partition assignment (if enabled), and to use for fetching and committing offsets. |
fetch_max_wait_ms | number | Optional | The maximum amount of time in milliseconds the server will block before answering the fetch request if there isn’t sufficient data to immediately satisfy the requirement given by fetch_min_bytes. Default is 500. |
fetch_min_bytes | number | Optional | Minimum amount of data the server should return for a fetch request, otherwise wait up to fetch_max_wait_ms for more data to accumulate. Default is 1. |
fetch_max_bytes | number | Optional | The maximum amount of data the server should return for a fetch request. |
max_partition_fetch_bytes | number | Optional | The maximum amount of data per-partition the server will return. |
request_timeout_ms | number | Optional | Client request timeout in milliseconds. Default is 305000. |
retry_backoff_ms | number | Optional | Milliseconds to backoff when retrying on errors. Default is 100. |
reconnect_backoff_ms | number | Optional | The amount of time in milliseconds to wait before attempting to reconnect to a given host. Default is 50. |
reconnect_backoff_max_ms | number | Optional | The maximum amount of time in milliseconds to backoff/wait when reconnecting to a broker that has repeatedly failed to connect. |
max_in_flight_requests_per_connection | number | Optional | Requests are pipelined to kafka brokers up to this number of maximum requests per broker connection. Note that if this setting is set to be greater than 1 and there are failed sends, there is a risk of message re-ordering due to retries (i.e., if retries are enabled). Default is 5 |
auto_offset_reset | string | Optional | A policy for resetting offsets on OffsetOutOfRange errors are earliest will move to the oldest available message, ‘latest’ will move to the most recent. Any other value will raise the exception. Default is ‘latest’. |
enable_auto_commit | boolean | Optional | If True , the consumer’s offset will be periodically committed in the background. Default is True. |
auto_commit_interval_ms | number | Optional | Number of milliseconds between automatic offset commits, if enable_auto_commit is True. Default is 5000. |
check_crcs | boolean | Optional | Automatically check the CRC32 of the records consumed. Default is True |
metadata_max_age_ms | number | Optional | The period of time in milliseconds after which we force a refresh of metadata, even if we haven’t seen any partition leadership changes to proactively discover any new brokers or partitions. Default is 300000 |
max_poll_records | number | Optional | The maximum number of records returned in a single call to poll(). Default is 500 |
max_poll_interval_ms | number | Optional | The maximum delay between invocations of poll() when using consumer group management. |
session_timeout_ms | number | Optional | The timeout used to detect failures when using Kafka’s group management facilities. |
heartbeat_interval_ms | number | Optional | The expected time in milliseconds between heartbeats to the consumer coordinator when using Kafka’s group management facilities. |
Input Example
{"topic":"first_topic","partitions":"1,2","partition_offsets":{"1":3,"2":4},"limit":100,"return_single_record":true,"client_id":"kafka-python-","group_id":1,"fetch_max_wait_ms":500,"fetch_min_bytes":1,"fetch_max_bytes":52428800,"max_partition_fetch_bytes":1048576,"request_timeout_ms":305000,"retry_backoff_ms":100,"reconnect_backoff_ms":50,"reconnect_backoff_max_ms":1000,"max_in_flight_requests_per_connection":5,"auto_offset_reset":"latest","enable_auto_commit":true,"auto_commit_interval_ms":5000,"check_crcs":true,"metadata_max_age_ms":300000,"max_poll_records":500,"max_poll_interval_ms":300000,"session_timeout_ms":10000,"heartbeat_interval_ms":3000,"receive_buffer_bytes":32768,"send_buffer_bytes":131072,"consumer_timeout_ms":1000,"security_protocol":"PLAINTEXT","api_version":"(0,9)","api_version_auto_timeout_ms":2000,"connections_max_idle_ms":540000,"metrics_num_samples":2,"metrics_sample_window_ms":30000,"exclude_internal_topics":true,"sasl_mechanism":"PLAIN","sasl_kerberos_service_name":"kafka","sasl_kerberos_domain_name":"localhost"}
Output
Parameter | Type | Description |
|---|---|---|
status_code | number | HTTP status code of the response |
reason | string | Response reason phrase |
messages | array | Response message |
messages.message | string | Response message |
partition_offsets | object | Output field: partition_offsets |
partition_offsets.1 | number | Output field: partition_offsets.1 |
partition_offsets.2 | number | Output field: partition_offsets.2 |
Output Example
{"status_code":200,"response_headers":{},"reason":"OK","json_body":{"messages":[{},{}],"partition_offsets":{"1":5,"2":10}}}
Send Data
Sends Data to a Kafka Topic.
Input
Argument Name | Type | Required | Description |
|---|---|---|---|
topic | string | Required | Kafka Topic. |
data | string | Required | Data to send over Kafka. |
client_id | string | Optional | A name for this client. |
acks | number | Optional | The number of acknowledgments the producer requires the leader to have received before considering a request complete. |
compression_type | string | Optional | The compression type for all data generated by the producer. Valid values are ‘gzip’, ‘snappy’, ‘lz4’, or None. |
retries | number | Optional | Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error. |
batch_size | number | Optional | Requests sent to brokers will contain multiple batches, one for each partition with data available to be sent. A small batch size will make batching less common and may reduce throughput (a batch size of zero will disable batching entirely). Default is 16384 |
linger_ms | number | Optional | The producer groups together any records that arrive in between request transmissions into a single batched request. Default is 0 |
buffer_memory | number | Optional | The total bytes of memory the producer should use to buffer records waiting to be sent to the server. |
connections_max_idle_ms | number | Optional | Close idle connections after the number of milliseconds specified by this config. |
max_block_ms | number | Optional | Number of milliseconds to block during send() and partitions_for() |
max_request_size | number | Optional | The maximum size of a request. |
metadata_max_age_ms | number | Optional | The period of time in milliseconds after which we force a refresh of metadata even if we haven’t seen any partition leadership changes to proactively discover any new brokers or partitions. |
retry_backoff_ms | number | Optional | Milliseconds to backoff when retrying on errors. Default is 100 |
request_timeout_ms | number | Optional | Client request timeout in milliseconds. Default is 30000. |
receive_buffer_bytes | number | Optional | The size of the TCP receive buffer (SO_RCVBUF) to use when reading data. |
send_buffer_bytes | number | Optional | The size of the TCP send buffer (SO_SNDBUF) to use when sending data. |
reconnect_backoff_ms | number | Optional | The amount of time in milliseconds to wait before attempting to reconnect to a given host. Defaults to 50. |
reconnect_backoff_max_ms | number | Optional | The amount of time in milliseconds to wait before attempting to reconnect to a given host. Default is 50. |
max_in_flight_requests_per_connection | number | Optional | Requests are pipelined to kafka brokers up to this number of maximum requests per broker connection. Note that if this setting is set to be greater than 1 and there are failed sends, there is a risk of message re-ordering due to retries (i.e., if retries are enabled). Default is 5 |
security_protocol | string | Optional | Protocol used to communicate with brokers. Valid values are PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL. Default is PLAINTEXT. |
api_version | string | Optional | Specify which Kafka API version to use. |
api_version_auto_timeout_ms | number | Optional | number of milliseconds to throw a timeout exception from the constructor when checking the broker api version. |
metrics_num_samples | number | Optional | The number of samples maintained to compute metrics. Default is 2 |
metrics_sample_window_ms | number | Optional | The maximum age in milliseconds of samples used to compute metrics. Default is 30000 |
Input Example
{"topic":"first_topic","data":"test data","client_id":"kafka-python-producer-#","compression_type":"gzip","retries":0,"batch_size":16384,"linger_ms":0,"buffer_memory":33554432,"connections_max_idle_ms":540000,"max_block_ms":60000,"max_request_size":1048576,"metadata_max_age_ms":300000,"retry_backoff_ms":100,"request_timeout_ms":30000,"receive_buffer_bytes":32768,"send_buffer_bytes":131072,"reconnect_backoff_ms":50,"reconnect_backoff_max_ms":1000,"max_in_flight_requests_per_connection":5,"security_protocol":"PLAINTEXT","api_version":"(0,9)","api_version_auto_timeout_ms":2000,"metrics_num_samples":2,"metrics_sample_window_ms":30000,"sasl_mechanism":"PLAIN","sasl_kerberos_service_name":"kafka","sasl_kerberos_domain_name":"localhost"}
Output
Parameter | Type | Description |
|---|---|---|
status_code | number | HTTP status code of the response |
reason | string | Response reason phrase |
success | boolean | Whether the operation was successful |
Output Example
{"status_code":200,"response_headers":{},"reason":"OK","json_body":{"success":true}}
Response Headers
Header | Description | Example |
|---|---|---|
Content-Type | The media type of the resource | application/json |
Date | The date and time at which the message was originated | Thu, 01 Jan 2024 00:00:00 GMT |