Managing connectors
Connectors manage the transfer of Apache Kafka® topics to other clusters or data storage systems.
You can:
- Get a list of connectors.
- Get detailed information about a connector.
- Create a connector.
- Edit a connector.
- Pause a connector.
- Resume a connector.
- Import a connector to Terraform.
- Delete a connector.
Getting a list of connectors
- In the management console
, go to the appropriate folder. - In the list of services, select Managed Service for Kafka.
- Select a cluster and open the Connectors tab.
If you do not have the Yandex Cloud command line interface yet, install and initialize it.
The folder specified in the CLI profile is used by default. You can specify a different folder using the --folder-name
or --folder-id
parameter.
To get a list of cluster connectors, run the command:
yc managed-kafka connector list --cluster-name=<cluster_name>
Result:
+--------------+-----------+
| NAME | TASKS MAX |
+--------------+-----------+
| connector559 | 1 |
| ... | |
+--------------+-----------+
You can retrieve the cluster name with a list of clusters in the folder.
To get a list of connectors, use the list REST API method for the Connector resource or the ConnectorService/List gRPC API call and provide the cluster ID in the clusterId
request parameter.
To find out the cluster ID, get a list of clusters in the folder.
Getting detailed information about a connector
- In the management console
, go to the appropriate folder. - In the list of services, select Managed Service for Kafka.
- Select a cluster and open the Connectors tab.
- Click the name of the connector you need.
If you do not have the Yandex Cloud command line interface yet, install and initialize it.
The folder specified in the CLI profile is used by default. You can specify a different folder using the --folder-name
or --folder-id
parameter.
To get detailed information about a connector, run this command:
yc managed-kafka connector get <connector_name>\
--cluster-name=<cluster_name>
Result:
name: connector785
tasks_max: "1"
cluster_id: c9qbkmoiimsl********
...
You can request the connector name with a list of cluster connectors and the cluster name with a list of clusters in the folder.
To get connector details, use the get REST API method for the Connector resource or the ConnectorService/Get gRPC API call and provide the following in the request:
- Cluster ID in the
clusterId
parameter. To find out the cluster ID, get a list of clusters in the folder. - Connector name in the
connectorName
parameter. To find out the name, retrieve a list of cluster connectors.
Creating a connector
-
In the management console
, go to the appropriate folder. -
In the list of services, select Managed Service for Kafka.
-
Select a cluster and open the Connectors tab.
-
Click Create connector.
-
Under Basic parameters, specify:
- Connector name.
- Task limit: Number of concurrent processes. We recommend a value of at least
2
for even replication load distribution.
-
Under Additional properties, specify the connector properties in the following format:
<key>:<value>
The key can either be a simple string or contain a prefix indicating that it belongs to the source or target (a cluster alias in the connector configuration):
<cluster_alias>.<key_body>:<value>
-
Select the connector type: MirrorMaker or S3 Sink.
-
Specify a configuration for the selected connector.
For more information about the supported connector types, see Connectors.
-
Click Create.
If you do not have the Yandex Cloud command line interface yet, install and initialize it.
The folder specified in the CLI profile is used by default. You can specify a different folder using the --folder-name
or --folder-id
parameter.
To create a MirrorMaker connector:
-
View a description of the CLI command to create a connector:
yc managed-kafka connector-mirrormaker create --help
-
Create a connector:
yc managed-kafka connector-mirrormaker create <connector_name> \ --cluster-name=<cluster_name> \ --direction=<connector_direction> \ --tasks-max=<task_limit> \ --properties=<advanced_properties> \ --replication-factor=<replication_factor> \ --topics=<topic_template> \ --this-cluster-alias=<prefix_for_this_cluster> \ --external-cluster alias=<prefix_for_external_cluster>,` `bootstrap-servers=<list_of FQDNs_of_broker_hosts>,` `security-protocol=<security_protocol>,` `sasl-mechanism=<encryption_mechanism>,` `sasl-username=<username>,` `sasl-password=<user_password>,` `ssl-truststore-certificates=<certificates_in_PEM_format>
Learn how to get the FQDN of a broker host in this guide.
You can retrieve the cluster name with a list of clusters in the folder.
The
--direction
parameter takes the value:egress
: If the current cluster is a source cluster.ingress
: If the current cluster is a target cluster.
To create an S3 Sink connector:
-
View a description of the CLI command to create a connector:
yc managed-kafka connector-s3-sink create --help
-
Create a connector:
yc managed-kafka connector-s3-sink create <connector_name> \ --cluster-name=<cluster_name> \ --tasks-max=<task_limit> \ --properties=<advanced_properties> \ --topics=<topic_template> \ --file-compression-type=<compression_codec> \ --file-max-records=<maximum_number_of_messages_per_file> \ --bucket-name=<bucket_name> \ --access-key-id=<ID_of_AWS-compatible_static_key> \ --secret-access-key=<contents_of_AWS-compatible_static_key> \ --storage-endpoint=<endpoint_of_S3-compatible_storage> \ --region=<region_of_S3-compatible_storage>
You can retrieve the cluster name with a list of clusters in the folder.
-
Check the list of Mirrormaker and S3 Sink connector settings.
-
Open the current Terraform configuration file with an infrastructure plan.
For more information about how to create this file, see Creating clusters.
-
To create a Mirrormaker connector, add the
yandex_mdb_kafka_connector
resource with theconnector_config_mirrormaker
settings section:resource "yandex_mdb_kafka_connector" "<connector_name>" { cluster_id = "<cluster_ID>" name = "<connector_name>" tasks_max = <task_limit> properties = { <advanced_properties> } connector_config_mirrormaker { topics = "<topic_template>" replication_factor = <replication_factor> source_cluster { alias = "<prefix_for_the_cluster>" external_cluster { bootstrap_servers = "<list_of_FQDNs_of_broker_hosts>" sasl_username = "<username>" sasl_password = "<user_password>" sasl_mechanism = "<encryption_mechanism>" security_protocol = "<security_protocol>" ssl-truststore-certificates = "<contents_of_PEM_certificate>" } } target_cluster { alias = "<prefix_for_the_cluster>" this_cluster {} } } }
Learn how to get the FQDN of a broker host in this guide.
-
To create an S3 Sink connector, add the
yandex_mdb_kafka_connector
resource with theconnector_config_s3_sink
settings section:resource "yandex_mdb_kafka_connector" "<connector_name>" { cluster_id = "<cluster_ID>" name = "<connector_name>" tasks_max = <task_limit> properties = { <advanced_properties> } connector_config_s3_sink { topics = "<topic_template>" file_compression_type = "<compression_codec>" file_max_records = <maximum_number_of_messages_per_file> s3_connection { bucket_name = "<bucket_name>" external_s3 { endpoint = "<endpoint_of_S3-compatible_storage>" access_key_id = "<ID_of_AWS-compatible_static_key>" secret_access_key = "<contents_of_AWS_compatible_static_key>" } } } }
-
Make sure the settings are correct.
-
Using the command line, navigate to the folder that contains the up-to-date Terraform configuration files with an infrastructure plan.
-
Run the command:
terraform validate
If there are errors in the configuration files, Terraform will point to them.
-
-
Confirm updating the resources.
-
Run the command to view planned changes:
terraform plan
If the resource configuration descriptions are correct, the terminal will display a list of the resources to modify and their parameters. This is a test step. No resources are updated.
-
If you are happy with the planned changes, apply them:
-
Run the command:
terraform apply
-
Confirm the update of resources.
-
Wait for the operation to complete.
-
-
For more information, see the Terraform provider documentation
To create a connector, use the create REST API method for the Connector resource or the ConnectorService/Create gRPC API call and provide the following in the request:
- In the
clusterId
parameter, the ID of the cluster where you want to create a connector. To find out the cluster ID, get a list of clusters in the folder. - Connector settings in the
connectorSpec
parameter.
Editing a connector
- In the management console
, go to the appropriate folder. - In the list of services, select Managed Service for Kafka.
- Select a cluster and open the Connectors tab.
- In the line with the required connector, click
- Edit the connector properties as needed.
- Click Save.
If you do not have the Yandex Cloud command line interface yet, install and initialize it.
The folder specified in the CLI profile is used by default. You can specify a different folder using the --folder-name
or --folder-id
parameter.
To edit a MirrorMaker connector:
-
View a description of the CLI command to edit a connector:
yc managed-kafka connector-mirrormaker update --help
-
Run an operation, such as the task limit change operation:
yc managed-kafka connector-mirrormaker update <connector_name> \ --cluster-name=<cluster_name> \ --direction=<connector_direction> \ --tasks-max=<new_task_limit>
Where
--direction
is the connector direction:ingress
oregress
.You can request the connector name with a list of cluster connectors and the cluster name with a list of clusters in the folder.
To update the S3 Sink connector:
-
View a description of the CLI command to edit a connector:
yc managed-kafka connector-s3-sink update --help
-
Run an operation, such as the task limit change operation:
yc managed-kafka connector-s3-sink update <connector_name> \ --cluster-name=<cluster_name> \ --tasks-max=<new_task_limit>
You can request the connector name with a list of cluster connectors and the cluster name with a list of clusters in the folder.
-
Check the list of Mirrormaker and S3 Sink connector settings.
-
Open the current Terraform configuration file with an infrastructure plan.
For more information about how to create this file, see Creating clusters.
-
Edit the parameter values in the
yandex_mdb_kafka_connector
resource description:-
For the Mirrormaker connector:
resource "yandex_mdb_kafka_connector" "<connector_name>" { cluster_id = "<cluster_ID>" name = "<connector_name>" tasks_max = <task_limit> properties = { <advanced_properties> } connector_config_mirrormaker { topics = "<topic_template>" replication_factor = <replication_factor> source_cluster { alias = "<prefix_for_the_cluster>" external_cluster { bootstrap_servers = "<list_of_FQDNs_of_broker_hosts>" sasl_username = "<username>" sasl_password = "<user_password>" sasl_mechanism = "<encryption_mechanism>" security_protocol = "<security_protocol>" ssl-truststore-certificates = "<contents_of_PEM_certificate>" } } target_cluster { alias = "<prefix_for_the_cluster>" this_cluster {} } } }
-
For the S3 Sink connector:
resource "yandex_mdb_kafka_connector" "<S3_Sink_connector_name>" { cluster_id = "<cluster_ID>" name = "<S3_Sink_connector_name>" tasks_max = <task_limit> properties = { <advanced_properties> } connector_config_s3_sink { topics = "<topic_template>" file_max_records = <maximum_number_of_messages_per_file> s3_connection { bucket_name = "<bucket_name>" external_s3 { endpoint = "<endpoint_of_S3-compatible_storage>" access_key_id = "<ID_of_AWS_compatible_static_key>" secret_access_key = "<contents_of_AWS_compatible_static_key>" } } } }
-
-
Make sure the settings are correct.
-
Using the command line, navigate to the folder that contains the up-to-date Terraform configuration files with an infrastructure plan.
-
Run the command:
terraform validate
If there are errors in the configuration files, Terraform will point to them.
-
-
Confirm updating the resources.
-
Run the command to view planned changes:
terraform plan
If the resource configuration descriptions are correct, the terminal will display a list of the resources to modify and their parameters. This is a test step. No resources are updated.
-
If you are happy with the planned changes, apply them:
-
Run the command:
terraform apply
-
Confirm the update of resources.
-
Wait for the operation to complete.
-
-
For more information, see the Terraform provider documentation
To update a connector, use the update REST API method for the Connector resource or the ConnectorService/Update gRPC API call and provide the following in the request:
- ID of the cluster where you wish to create a connector in the
clusterId
parameter. To find out the cluster ID, get a list of clusters in the folder. - Connector settings in the
connectorSpec
parameter.
Pausing a connector
When you pause a connector:
- The connection to the target is broken.
- Data is deleted from the connector service topics.
To pause a connector:
- In the management console
, go to the appropriate folder. - In the list of services, select Managed Service for Kafka.
- Select a cluster and open the Connectors tab.
- Click
If you do not have the Yandex Cloud command line interface yet, install and initialize it.
The folder specified in the CLI profile is used by default. You can specify a different folder using the --folder-name
or --folder-id
parameter.
To pause a connector, run the command:
yc managed-kafka connector pause <connector_name> \
--cluster-name=<cluster_name>
To pause a connector, use the pause REST API method for the Connector resource or the ConnectorService/Pause gRPC API call and provide the following in the request:
- Cluster ID in the
clusterId
parameter. To find out the cluster ID, get a list of clusters in the folder. - Connector name in the
connectorName
parameter. To find out the name, retrieve a list of cluster connectors.
Resuming a connector
- In the management console
, go to the appropriate folder. - In the list of services, select Managed Service for Kafka.
- Select a cluster and open the Connectors tab.
- Click
If you do not have the Yandex Cloud command line interface yet, install and initialize it.
The folder specified in the CLI profile is used by default. You can specify a different folder using the --folder-name
or --folder-id
parameter.
To resume a connector, run the command:
yc managed-kafka connector resume <connector_name> \
--cluster-name=<cluster_name>
To resume a connector, use the resume REST API method for the Connector resource or the ConnectorService/Resume gRPC API call and provide the following in the request:
- Cluster ID in the
clusterId
parameter. To find out the cluster ID, get a list of clusters in the folder. - Connector name in the
connectorName
parameter. To find out the name, retrieve a list of cluster connectors.
Importing a connector to Terraform
Using import, you can bring the existing connectors under Terraform management.
-
In the Terraform configuration file, specify the connector you want to import:
resource "yandex_mdb_kafka_cluster" "<connector_name>" {}
-
Run the following command to import the connector:
terraform import yandex_mdb_kafka_connector.<connector_name> <cluster_ID>:<connector_name>
To learn more about importing connectors, see the Terraform provider documentation
.
Deleting a connector
- In the management console
, go to the appropriate folder. - In the list of services, select Managed Service for Kafka.
- Select a cluster and open the Connectors tab.
- Click
- Click Delete.
If you do not have the Yandex Cloud command line interface yet, install and initialize it.
The folder specified in the CLI profile is used by default. You can specify a different folder using the --folder-name
or --folder-id
parameter.
To delete a connector, run the command:
yc managed-kafka connector delete <connector_name> \
--cluster-name <cluster_name>
-
Open the current Terraform configuration file with an infrastructure plan.
For more information about how to create this file, see Creating clusters.
-
Delete the
yandex_mdb_kafka_connector
resource with the description of the connector you need. -
Make sure the settings are correct.
-
Using the command line, navigate to the folder that contains the up-to-date Terraform configuration files with an infrastructure plan.
-
Run the command:
terraform validate
If there are errors in the configuration files, Terraform will point to them.
-
-
Confirm updating the resources.
-
Run the command to view planned changes:
terraform plan
If the resource configuration descriptions are correct, the terminal will display a list of the resources to modify and their parameters. This is a test step. No resources are updated.
-
If you are happy with the planned changes, apply them:
-
Run the command:
terraform apply
-
Confirm the update of resources.
-
Wait for the operation to complete.
-
-
For more information, see the Terraform provider documentation
To delete a connector, use the delete REST API method for the Connector resource or the ConnectorService/Delete gRPC API call and provide the following in the request:
- Cluster ID in the
clusterId
parameter. To find out the cluster ID, get a list of clusters in the folder. - Connector name in the
connectorName
parameter. To find out the name, retrieve a list of cluster connectors.
Connector parameters
MirrorMaker
-
Topics: Template for selecting topics to replicate. Topic names are separated by a comma or
|
in the list. You can use the.*
expression, for example,analysis.*
. To migrate all topics, specify.*
. -
Replication factor: Number of topic copies stored in the cluster.
-
Under Source cluster, specify the parameters for connecting to the source cluster:
-
Alias: Prefix for the source cluster in the connector settings.
Note
Topics in the target cluster are created with the indicated prefix.
-
Use this cluster: Select this option to use the current cluster as a source.
-
Bootstrap servers: Comma-separated list of the FQDNs of the source cluster broker hosts with the port numbers to connect to, e.g.,
broker1.example.com:9091,broker2.example.com
.Learn how to get the FQDN of a broker host in this guide.
-
SASL username: Username for connecting the connector to the source cluster.
-
SASL password: User password for connecting the connector to the source cluster.
-
SASL mechanism: Select a mechanism for name and password encryption.
-
Security protocol: Select a protocol for connecting the connector:
plaintext
,sasl_plaintext
: For non-SSL connectionsssl
,sasl_ssl
: For SSL connections
-
Certificate in PEM format: Upload a PEM certificate to access the external cluster.
-
-
Under Target cluster, specify the parameters for connecting to the target cluster:
-
Alias: Prefix for the target cluster in the connector settings.
-
Use this cluster: Select this option to use the current cluster as a target.
-
Bootstrap servers: Сomma-separated list of the FQDNs of the target cluster broker hosts with the port numbers to connect to.
Learn how to get the FQDN of a broker host in this guide.
-
SASL username: Username for connecting the connector to the target cluster.
-
SASL password: User password for connecting the connector to the target cluster.
-
SASL mechanism: Select a mechanism for name and password encryption.
-
Security protocol: Select a protocol for connecting the connector:
plaintext
,sasl_plaintext
: For non-SSL connectionsssl
,sasl_ssl
: For SSL connections
-
Certificate in PEM format: Upload a PEM certificate to access the external cluster.
-
-
To specify additional setting values not listed above, create the relevant keys and specify their values under Additional properties when creating or editing a connector. Here are some sample keys:
key.converter
value.converter
For the list of common connector settings, see the Apache Kafka® documentation
.
-
--cluster-name
: Cluster name. -
--direction
: Connector direction:ingress
: If the cluster is a target.egress
: If the cluster is a source.
-
--tasks-max
: Number of concurrent processes. We recommend a value of at least2
for even replication load distribution. -
--properties
: Comma-separated list of advanced connector settings in<key>:<value>
format. Here are some sample keys:key.converter
value.converter
For the list of common connector settings, see the Apache Kafka® documentation
. -
--replication-factor
: Number of topic copies stored in the cluster. -
--topics
: Template for selecting topics to replicate. Topic names are separated by a comma or|
in the list. You can use the.*
expression, for example,analysis.*
. To migrate all topics, specify.*
. -
--this-cluster-alias
: Prefix for this cluster in the connector settings. -
--external-cluster
: Parameters of an external cluster:-
alias
: Prefix for the external cluster in the connector settings. -
bootstrap-servers
: Comma-separated list of the FQDNs of the external cluster broker hosts with the port numbers to connect to.Learn how to get the FQDN of a broker host in this guide.
-
security-protocol
: Connector connection protocol:plaintext
,sasl_plaintext
: For non-SSL connectionsssl
,sasl_ssl
: For SSL connections
-
sasl-mechanism
: Name and password encryption mechanism. -
sasl-username
: Username for connecting the connector to the external cluster. -
sasl-password
: User password for connecting the connector to the external cluster. -
ssl-truststore-certificates
: List of PEM certificates.
-
-
properties: Comma-separated list of advanced connector settings in
<key>:<value>
format. Here are some sample keys:key.converter
value.converter
For the list of common connector settings, see the Apache Kafka® documentation
. -
topics: Template for selecting topics to replicate. Topic names are separated by a comma or
|
in the list. You can use the.*
expression, for example,analysis.*
. To migrate all topics, specify.*
. -
replication_factor: Number of topic copies stored in the cluster.
-
source_cluster and target_cluster: Parameters for connecting to the source cluster and target cluster:
-
alias: Prefix for the cluster in the connector settings.
Note
Topics in the target cluster are created with the indicated prefix.
-
this_cluster: Option to use the current cluster as a source or target.
-
external_cluster: Parameters for connecting to the external cluster:
-
bootstrap_servers: Comma-separated list of the FQDNs of the cluster broker hosts with the port numbers to connect to.
Learn how to get the FQDN of a broker host in this guide.
-
sasl_username: Username for connecting the connector to the cluster.
-
sasl_password: User password for connecting the connector to the cluster.
-
sasl_mechanism: Name and password encryption mechanism.
-
security_protocol: Connector connection protocol:
plaintext
,sasl_plaintext
: For non-SSL connectionsssl
,sasl_ssl
: For SSL connections
-
ssl_truststore_certificates: PEM certificate contents.
-
-
S3 Sink
-
Topics: Template for selecting topics to replicate. Topic names are separated by a comma or
|
in the list. You can use the.*
expression, for example,analysis.*
. To migrate all topics, specify.*
. -
Compression type: Select the codec to compress messages:
You cannot change this parameter after creating the cluster.
-
(Optional) Max record per file: Maximum number of records that can be written to a single file in an S3-compatible storage.
-
Under S3 connection, specify the storage connection parameters:
-
Bucket: Storage bucket name
-
Endpoint: Endpoint for storage access (to be requested from the storage provider)
-
(Optional) Region: Region name. Default:
us-east-1
. Available regions . -
(Optional) Access key ID, Secret access key: ID and contents of the AWS compatible key.
-
-
To specify additional setting values not listed above, create the relevant keys and specify their values under Additional properties when creating or editing a connector. Here are some sample keys:
key.converter
value.converter
value.converter.schemas.enable
format.output.type
For the list of all connector settings, see the connector documentation
. For the list of common connector settings, see the Apache Kafka® documentation .
-
--cluster-name
: Cluster name. -
--tasks-max
: Number of concurrent processes. We recommend a value of at least2
for even replication load distribution. -
--properties
: Comma-separated list of advanced connector settings in<key>:<value>
format. Here are some sample keys:key.converter
value.converter
value.converter.schemas.enable
format.output.type
For the list of all connector settings, see the connector documentation
. For the list of common connector settings, see the Apache Kafka® documentation . -
--topics
: Template for selecting topics to replicate. Topic names are separated by a comma or|
in the list. You can use the.*
expression, for example,analysis.*
. To migrate all topics, specify.*
. -
--file-compression-type
: Codec for message compression. You cannot change this parameter after creating the cluster. Acceptable values include: -
--file-max-records
: Maximum number of records that can be written to a single file in S3-compatible storage. -
--bucket-name
: Name of the bucket in S3-compatible storage to write data to. -
--storage-endpoint
: Endpoint for storage access (to find out from storage provider). Example:storage.yandexcloud.net
. -
--region
: Region where the bucket of S3-compatible storage is located. Default:us-east-1
. Available regions . -
--access-key-id
,--secret-access-key
: AWS-compatible key ID and contents.
-
properties: Comma-separated list of advanced connector settings in
<key>:<value>
format. Here are some sample keys:key.converter
value.converter
value.converter.schemas.enable
format.output.type
For the list of all connector settings, see the connector documentation
. For the list of common connector settings, see the Apache Kafka® documentation . -
topics: Template for selecting topics to replicate. Topic names are separated by a comma or
|
in the list. You can use the.*
expression, for example,analysis.*
. To migrate all topics, specify.*
. -
file_compression_type: Codec for message compression. You cannot change this parameter after creating the cluster. Acceptable values include:
-
file_max_records: Maximum number of records that can be written to a single file in S3-compatible storage.
-
s3_connection: S3-compatible storage connection parameters:
-
bucket_name: Name of the bucket to write data to.
-
external_s3: External S3-compatible storage connection parameters:
-
endpoint: Endpoint for storage access (to find out from storage provider). Example:
storage.yandexcloud.net
. -
region: Region where the bucket of S3-compatible storage is located. Default:
us-east-1
. Available regions . -
access_key_id, secret_access_key: AWS-compatible key ID and contents.
-
-