Before creating a kafka connect, it’s highly recommended to do the following:
Check kafka connect availability to ensure the version number of the kafka connect you want to create is supported on StreamNative Cloud.
Go to StreamNative Hub and find the connect-specific docs of your version for configuration reference.
You may see below error logs for the first time you create a connector:
org.apache.kafka.common.config.ConfigException: Topic '__kafka_connect_offset_storage' supplied via the 'offset.storage.topic' property is required to have 'cleanup.policy=compact' to guarantee consistency and durability of source connector offsets, but found the topic currently has 'cleanup.policy=delete'. Continuing would likely result in eventually losing source connector offsets and problems restarting this Connect cluster in the future. Change the 'offset.storage.topic' property in the Connect worker configurations to use a topic with 'cleanup.policy=compact'.
You should set the cleanup.policy of the __kafka_connect_offset_storage topic to compact to avoid the above error with below command:
In the left navigation pane, click Connectors, then click the Kafka Sources tab.
Click the Create Kafka Source button:
Fill in the required fields and optional fields as you wish, and then Click the Submit button.
If you want to verify whether the data generator source connect has been created successfully, go back to the Connectors page, and you should see the created connector in the Kafka Sources tab, like below:
StreamNative Cloud supports Single Message Transformations (SMTs) for Kafka Connect. You can use SMTs to transform messages before they are written to the target system.The following example shows how to create a Datagen source connector named test on StreamNative Cloud using different tools.Please refer to the Kafka Connect SMTs to check the supported SMTs in StreamNative cloud.
snctl
kcctl
Rest API
Console
To create a data generator source connect named test with SMT, run the following command.
In the left navigation pane, click Connectors, then click the Kafka Sources tab.
Click the Create Kafka Source button:
Fill in the required fields and optional fields as you wish
Click the Advance Settings tab, and then fill in the SMT fields:
Click the Submit button.
If you want to verify whether the data generator source connect has been created successfully, go back to the Connectors page, and you should see the created connector in the Kafka Sources tab, like below:
Some connects require sensitive information, such as passwords, token, to be passed to the connector. And you may not want to expose these sensitive information in the connector configuration.
To solve this problem, you can use the following methods to pass sensitive information to the connector:
Create a secretFor example, the Milvus sink connector requires a token to be passed to the connector.
You can create a secret in the console UI and pass the secret name to the connector configuration.
The location should be the same as the region of your Pulsar cluster.The awsAccessKey and awsSecretKey is the field name, and the lambda-sink-secret can be any unique name you want to give to the secret.For a Milvus sink, we should create a secret with a token field.
Pass secrets to the connector configurationThe following example shows how to create a Milvus sink connector named test on StreamNative Cloud using different tools.
snctl
kcctl
Rest API
Console
To create a Milvus sink connector named test, run the following command.
In the left navigation pane, click Connectors, then click the Kafka Sinks tab.
Click the Create Kafka Sink button, and then choose the Milvus connect:
In the Authentication Secrets selection box, you can choose an existing secret or create a new secret.
Fill in the required fields and optional fields as you wish, and then Click the Submit button.
If you want to verify whether the data generator source connect has been created successfully, go back to the Connectors page, and you should see the created connector in the Kafka Sources tab, like below:
You can tune the Kafka client behavior for an individual connector by adding connector-level override settings. These settings are useful when one connector needs different throughput, latency, or retry behavior than the default Kafka Connect worker settings.Use the following prefixes in the connector configuration:
producer.override.*: Overrides Kafka producer settings. Use this prefix mainly for source connectors that write records to Kafka topics. Common settings include compression.type, batch.size, linger.ms, and acks.
consumer.override.*: Overrides Kafka consumer settings. Use this prefix for sink connectors that read records from Kafka topics. Common settings include max.poll.records, fetch.min.bytes, fetch.max.wait.ms, and auto.offset.reset.
admin.override.*: Overrides Kafka administrative client settings. Use this prefix when the connector needs different administrative client behavior, such as for topic creation or dead-letter queue operations. Common settings include request.timeout.ms, retry.backoff.ms, and default.api.timeout.ms.
For example, the following sink connector configuration increases the consumer batch size and adjusts admin client timeouts:
Kafka Connect supports using schema registry to save Avro/Protobuf/Json schema for the value and key. And StreamNative has an internal schema registry which can be used without complex configurations.
To use it, you just need to set the following configuration in the connector configuration:
value.converter.schema.registry.internal: true: if you want to use the internal schema registry for the value converter.
key.converter.schema.registry.internal: true: if you want to use the internal schema registry for the key converter.
Was this page helpful?
⌘I
Assistant
Responses are generated using AI and may contain mistakes.