Use the Confluent Cloud connector to send data to a Confluent Cloud cluster. If you are looking for information about how to create a connection to get data into Decodable from Confluent Cloud, see Confluent Cloud in the Connect to a data source chapter.

Confluent Cloud connections always use SSL transport encryption, which encrypts all network traffic between Decodable and the brokers.

Overview

Create a connection to Confluent Cloud with the Confluent Cloud connector

Follow these steps to send data from Decodable into Confluent Cloud.

1. From the Connections page, select the Confluent Cloud connector and complete the following fields.

2. Select the topic that you want Decodable to send data to. Select Next.

3. Select the stream that you’d like to connect to this connector. This is the stream containing the data that you'd like to send to Confluent Cloud. Select Next.

4. Define the connection’s schema. Decodable can auto-populate the connection’s schema using Confluent’s schema registry. In most cases, you’ll want to select Schema Registry to automatically populate the connection’s schema. However, if you would like to manually enter the schema, select New Schema or Structured Schema Definition.

   1. If you want to auto-populate the connection schema using Confluent’s schema registry, you’ll also need to provide the Schema Registry Endpoint, the Schema Registry API Key, and the Schema Registry API Secret to use to communicate with your schema registry.

5. Select Next when you are finished providing defining the connection’s schema.

6. Give the newly created connection a Name and Description and select Save.

Reference

Avro

A popular choice for data serialization on Kafka is Apache Avro . Decodable supports both plain Avro, as well as Avro with a schema registry.

Plain Avro

In order to use plain Avro, the connection property format=avro is required. In this mode, the Avro schema is generated from the connection schema, and is available on the inferred_schema connection property. For example:

./decodable connection get 2428ebe9
confluent_cloud_sink
  id                       2428ebe9
  description              -
  connector                confluent-cloud
  type                     sink
  stream id                0dc6bcbc
  fields
    0  request_id          STRING
    1  user_id             INT
    2  ts                  BIGINT
    3  endpoint            STRING
  primary key fields       -
  properties
    cluster.api.endpoint   https://example.us-west-2.aws.confluent.cloud:443
    cluster.api.key        NF2GERAAFNABCDEF
    cluster.api.secret     d5d7a670
    cluster.id             lkc-98765
    format                 avro
    inferred_schema        {"type":"record","name":"record","fields":[{"name":"request_id","type":["null","string"],"default":null},{"name":"user_id","type":["null","int"],"default":null},{"name":"ts","type":["null","long"],"default":null},{"name":"endpoint","type":["null","string"],"default":null}]}
    topic                  clickstream_topic
  target state             RUNNING
  actual state             RUNNING
  requested tasks          1
  actual tasks             1
  requested task size      M
  actual task size         -
  create time              2023-02-27T20:06:09Z
  update time              2023-02-27T20:06:09Z
  last runtime error       -

Avro with Schema Registry

To use Avro with a schema registry, the connection property format=confluent-avro is required. In this mode, the Avro schema is still derived from the connection schema, but it is validated and registered against the configured schema registry.

When using this mode, the following properties are used:

For example, you can create a source Kafka connection using both SASL Authentication, as well as a secured schema registry using:

decodable connection create \
--name avro_registry_source \
--description "A Kafka source connection with SASL auth and a schema registry" \
--type source \
--stream-id 295e2a7f \
--connector confluent-cloud \
--prop format="avro-confluent" \
--prop topic="my-topic" \
--prop bootstrap.servers="some.broker.cloud:9092" \
--prop security.protocol=SASL_SSL \
--prop sasl.username="my-api-key" \
--prop sasl.password="my-secret-key" \
--prop sasl.mechanism="PLAIN" \
--prop confluent-registry.url="https://my.schema.registry.cloud" \
--prop confluent-registry.api-key="<api-key>" \
--prop confluent-registry.api-secret="<api-secret>"