Skip to content

KafkaConnector

Usage example

Prerequisites

  • A Kubernetes cluster with the operator installed using helm, kubectl or kind (for local development).
  • A Kubernetes Secret with an Aiven authentication token.
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
apiVersion: aiven.io/v1alpha1
kind: Kafka
metadata:
  name: my-kafka
spec:
  authSecretRef:
    name: aiven-token
    key: token

  connInfoSecretTarget:
    name: kafka-secret

  project: my-aiven-project
  cloudName: google-europe-west1
  plan: business-4

  maintenanceWindowDow: friday
  maintenanceWindowTime: 23:00:00

  userConfig:
    kafka_connect: true
    kafka:
      group_max_session_timeout_ms: 70000
      log_retention_bytes: 1000000000

---

apiVersion: aiven.io/v1alpha1
kind: KafkaTopic
metadata:
  name: kafka-topic
spec:
  authSecretRef:
    name: aiven-token
    key: token

  project: my-aiven-project
  serviceName: my-kafka
  partitions: 3
  replication: 2

---

apiVersion: aiven.io/v1alpha1
kind: OpenSearch
metadata:
  name: my-os
spec:
  authSecretRef:
    name: aiven-token
    key: token

  connInfoSecretTarget:
    name: os-secret

  project: my-aiven-project
  cloudName: google-europe-west1
  plan: startup-4

  maintenanceWindowDow: friday
  maintenanceWindowTime: 23:00:00

---

apiVersion: aiven.io/v1alpha1
kind: KafkaConnector
metadata:
  name: my-kafka-connect
spec:
  authSecretRef:
    name: aiven-token
    key: token

  project: my-aiven-project
  serviceName: my-kafka
  connectorClass: io.aiven.kafka.connect.opensearch.OpensearchSinkConnector

  userConfig:
    topics:           my-kafka-topic
    type.name:        es-connector
    connection.url:   '{{ fromSecret "os-secret" "OPENSEARCH_URI" }}'

Apply the resource with:

kubectl apply -f example.yaml

Verify the newly created KafkaConnector:

kubectl get kafkaconnectors my-kafka-connect

The output is similar to the following: 

Name                Service Name    Project             Connector Class                                              State      Tasks Total            Tasks Running            
my-kafka-connect    my-kafka        my-aiven-project    io.aiven.kafka.connect.opensearch.OpensearchSinkConnector    RUNNING    <tasksStatus.total>    <tasksStatus.running>

KafkaConnector

KafkaConnector is the Schema for the kafkaconnectors API.

Required

  • apiVersion (string). Value aiven.io/v1alpha1.
  • kind (string). Value KafkaConnector.
  • metadata (object). Data that identifies the object, including a name string and optional namespace.
  • spec (object). KafkaConnectorSpec defines the desired state of KafkaConnector. See below for nested schema.

spec

Appears on KafkaConnector.

KafkaConnectorSpec defines the desired state of KafkaConnector.

Required

  • connectorClass (string, MaxLength: 1024). The Java class of the connector.
  • project (string, Immutable, Pattern: ^[a-zA-Z0-9_-]+$, MaxLength: 63). Identifies the project this resource belongs to.
  • serviceName (string, Immutable, Pattern: ^[a-z][-a-z0-9]+$, MaxLength: 63). Specifies the name of the service that this resource belongs to.
  • userConfig (object, AdditionalProperties: string). The connector-specific configuration To build config values from secret the template function {{ fromSecret "name" "key" }} is provided when interpreting the keys. Where "name" is the name of the secret and "key" is the key in the secret in the same namespace as the KafkaConnector.

Optional

authSecretRef

Appears on spec.

Authentication reference to Aiven token in a secret.

Required

  • key (string, MinLength: 1).
  • name (string, MinLength: 1).