Transferring data to an Apache Kafka® target endpoint

• Endpoint type: kafka_target.

• security_groups: Security groups for network traffic.

  Security group rules apply to a transfer. They allow opening up network access from the transfer VM to the cluster. For more information, see Networking in Yandex Data Transfer.

  Security groups must belong to the same network as the cluster.

  Note

  In Terraform, it is not required to specify a network for security groups.

• connection.cluster_id: ID of the cluster to connect to.

• auth: Authentication method used to connect to broker hosts:

  • sasl: SASL authentication.
  • no_auth: Without authentication.

Here is the configuration file example:

resource "yandex_datatransfer_endpoint" "<endpoint_name_in_Terraform>" {
  name = "<endpoint_name>"
  settings {
    kafka_target {
      security_groups = ["<list_of_security_group_IDs>"]
      connection {
        cluster_id = "<cluster_ID>"
      }
      auth {
        <authentication_method>
      }
      <topic_settings>
      <serialization_settings>
    }
  }
}

For more information, see the Terraform provider documentation.

• securityGroups: Security groups for network traffic, whose rules will apply to VMs and clusters without changing their settings. For more information, see Networking in Yandex Data Transfer.

• connection.clusterId: ID of the cluster to connect to.

• auth: Authentication method used to connect to broker hosts:

  • sasl: SASL authentication. The following parameters are required:

    • user: Name of the account Data Transfer will use to connect to the topic.
    • password.raw: Password for the account in text form.
    • mechanism: Hashing mechanism.

  • noAuth: Without authentication.

• Endpoint type: kafka_target.

• security_groups: Security groups for network traffic.

  Security group rules apply to a transfer. They allow opening up network access from the transfer VM to the broker hosts. For more information, see Networking in Yandex Data Transfer.

  Security groups must belong to the same network as the subnet_id subnet, if the latter is specified.

  Note

  In Terraform, it is not required to specify a network for security groups.

• connection.on_premise: Parameters for connecting to the broker hosts:

  • broker_urls: IP addresses or FQDNs of the broker hosts.

    If the Apache Kafka® port number differs from the standard one, specify it with a colon after the host name:

    <broker_host_IP_address_or_FQDN>:<port_number>
    

  • tls_mode: Parameters for encrypting the data to transfer, if required, e.g., for compliance with the PCI DSS requirements.

    • disabled: Disabled.
    • enabled: Enabled.

      • ca_certificate: CA certificate.

        Warning

        If no certificate is added, the transfer may fail with an error.

  • subnet_id: ID of the subnet hosting the broker hosts. The transfer will use this subnet to access them.

• auth: Authentication method used to connect to broker hosts:

  • sasl: SASL authentication.
  • no_auth: Without authentication.

Here is the configuration file example:

resource "yandex_datatransfer_endpoint" "<endpoint_name_in_Terraform>" {
  name = "<endpoint_name>"
  settings {
    kafka_target {
      security_groups = ["<list_of_security_group_IDs>"]
      connection {
        on_premise {
          broker_urls = ["<list_of_IP_addresses_or_broker_host_FQDNs>"]
          subnet_id  = "<ID_of_subnet_with_broker_hosts>"
        }
      }
      auth = {
        <authentication_method>
      }
      <topic_settings>
      <serialization_settings>
    }
  }
}

For more information, see the Terraform provider documentation.

• securityGroups: Security groups for network traffic, whose rules will apply to VMs and clusters without changing their settings. For more information, see Networking in Yandex Data Transfer.

• connection.onPremise: Parameters for connecting to the broker hosts:

  • brokerUrls: IP addresses or FQDNs of the broker hosts.

    If the Apache Kafka® port number differs from the standard one, specify it with a colon after the host name:

    <broker_host_IP_address_or_FQDN>:<port_number>
    

  • tlsMode: Parameters for encrypting the data to transfer, if required, e.g., for compliance with the PCI DSS requirements.

    • disabled: Disabled.
    • enabled: Enabled.

      • caCertificate: CA certificate.

        Warning

        If no certificate is added, the transfer may fail with an error.

  • subnetId: ID of the subnet hosting the broker hosts. The transfer will use this subnet to access them.

• auth: Authentication method used to connect to broker hosts:

  • sasl: SASL authentication. The following parameters are required:

    • user: Name of the account Data Transfer will use to connect to the topic.
    • password.raw: Password for the account in text form.
    • mechanism: Hashing mechanism.

  • noAuth: Without authentication.

Under topic_settings, specify one of the options to send messages to a topic:

• topic: Specify parameters in this section to send all messages to one topic:

  • topic_name: Name of the topic to send messages to.
  • save_tx_order: Option for saving the order of transactions. Set to true not to split the event stream into independent queues by table.

• topic_prefix: Specify a prefix to send messages to different topics with the given prefix.

  It is similar to the Debezium database.server.name setting. Messages will be sent to a topic named <topic_prefix>.<schema>.<table_name>.

In the topicSettings field, specify one of the options to send messages to a topic:

• topic: Specify parameters in this field to send all messages to one topic:

  • topicName: Name of the topic to send messages to.
  • saveTxOrder: Option for saving the order of transactions. Set to true not to split the event stream into independent queues by table.

• topicPrefix: Specify a prefix to send messages to different topics with the given prefix.

  It is similar to the Debezium database.server.name setting. Messages will be sent to a topic named <topic_prefix>.<schema>.<table_name>.

Under serializer, specify the selected serialization type:

• serializer_auto

• serializer_json

• serializer_debezium

  For this type, you can specify the Debezium serialization parameters as key/value pairs in the serializer_debezium.serializer_parameters field.

If you want to use JSON schemas in Yandex Schema Registry and preserve their compatibility when adding and deleting optional fields, add the serializer section with a description of the serialization settings to the configuration file. To generate a closed schema for keys, include the key.converter.dt.json.generate.closed.content.schema variable set to true in the serializer section. To generate a closed schema for values, include the value.converter.dt.json.generate.closed.content.schema variable set to true in the serializer section.

resource "yandex_datatransfer_endpoint" "<endpoint_name_in_Terraform>" {
  ...
  settings {
    kafka_target {
      ...
      serializer {
        serializer_debezium {
          serializer_parameters {            
            key = "key.converter.dt.json.generate.closed.content.schema"
            value = "true"
          }    
          serializer_parameters {            
            key = "value.converter.dt.json.generate.closed.content.schema"
            value = "true"
          }
          serializer_parameters {
            key = "value.converter.schemas.enable"
            value = "true"
          }
          serializer_parameters {
            key = "value.converter.schema.registry.url"
            value = "<Schema_Registry_namespace_URL>"
          }
          serializer_parameters {
            key = "value.converter.basic.auth.user.info"
            value = "api-key:<API_key_value>"
          }
        }
      }
    }
  }
}

Where:

• Schema_Registry_namespace_URL: Schema Registry namespace endpoint. You can copy the endpoint from the details for the Schema Registry namespace connection on the Debezium tab, in the value.converter.schema.registry.url parameter.

  Warning

  The namespace must have Compatibility check policy for JSON set to optional friendly.

• API_key_value: Value of the API key with a limited scope used for connecting to Schema Registry. To get this value:

  1. Create an API key with a limited scope and place it in the SECRET local variable:

     yc iam api-key create --folder-id <folder_ID> \
       --service-account-name <name_of_service_account_for_operations_with_Schema_Registry> \
       --scopes yc.schema-registry.schemas.manage \
       --expires-at '2030-01-01T00:00:00Z' >./api-key.yaml && \
     SECRET=`cat ./api-key.yaml | grep 'secret:' | awk '{print $2}'`
     

  2. Print the SECRET variable value to the terminal:

     echo $SECRET
     

  3. Copy the printed value and paste it into the value parameter in the configuration file.

Under serializer, specify the selected serialization type:

• serializerAuto

• serializerJson

• serializerDebezium

  For this type, you can specify the Debezium serialization parameters as key/value pairs in the serializerDebezium.serializerParameters field.

If you want to use JSON schemas in Yandex Schema Registry and preserve their compatibility when adding and deleting optional fields, add the serializer object with a description of the serialization settings to the request body. To generate a closed schema for keys, add the key.converter.dt.json.generate.closed.content.schema variable set to true in serializer. To generate a closed schema for values, add the value.converter.dt.json.generate.closed.content.schema variable set to true in serializer.

"serializer": {
  "serializerDebezium": {
    "serializerParameters": [
      {
        "key": "converter.dt.json.generate.closed.content.schema",
        "value": "true"
      },
      {
        "key": "value.converter",
        "value": "io.confluent.connect.json.JsonSchemaConverter"
      },
      {
        "key": "value.converter.schemas.enable",
        "value": "true"
      },
      {
        "key": "value.converter.schema.registry.url",
        "value": "<Schema_Registry_namespace_URL>"
      },
      {
        "key": "value.converter.basic.auth.user.info",
        "value": "api-key:<API_key_value>"
      }
    ]
  }
}

Where:

• Schema_Registry_namespace_URL: Schema Registry namespace endpoint. You can copy the endpoint from the details for the Schema Registry namespace connection on the Debezium tab, in the value.converter.schema.registry.url parameter.

  Warning

  The namespace must have Compatibility check policy for JSON set to optional friendly.

• API_key_value: Value of the API key with a limited scope used for connecting to Schema Registry. To get this value:

  1. Create an API key with a limited scope and place it in the SECRET local variable:

     yc iam api-key create --folder-id <folder_ID> \
       --service-account-name <name_of_service_account_for_operations_with_Schema_Registry> \
       --scopes yc.schema-registry.schemas.manage \
       --expires-at '2030-01-01T00:00:00Z' >./api-key.yaml && \
     SECRET=`cat ./api-key.yaml | grep 'secret:' | awk '{print $2}'`
     

  2. Print the SECRET variable value to the terminal:

     echo $SECRET
     

  3. Copy the printed value and paste it into the value parameter in the configuration file.