Chapter 3. Event sinks

Procedure

1. Create a KafkaSink object definition as a YAML file:

   Kafka sink YAML

   apiVersion: eventing.knative.dev/v1alpha1
   kind: KafkaSink
   metadata:
     name: <sink-name>
     namespace: <namespace>
   spec:
     topic: <topic-name>
     bootstrapServers:
      - <bootstrap-server>

   Copy to Clipboard Toggle word wrap

2. To create the Kafka sink, apply the KafkaSink YAML file:

   $ oc apply -f <filename>

   Copy to Clipboard Toggle word wrap

3. Configure an event source so that the sink is specified in its spec:

   Example of a Kafka sink connected to an API server source

   apiVersion: sources.knative.dev/v1alpha2
   kind: ApiServerSource
   metadata:
     name: <source-name> 

   1

   
     namespace: <namespace> 

   2

   
   spec:
     serviceAccountName: <service-account-name> 

   3

   
     mode: Resource
     resources:
     - apiVersion: v1
       kind: Event
     sink:
       ref:
         apiVersion: eventing.knative.dev/v1alpha1
         kind: KafkaSink
         name: <sink-name> 

   4

   Copy to Clipboard Toggle word wrap

   1

   The name of the event source.

   2

   The namespace of the event source.

   3

   The service account for the event source.

   4

   The Kafka sink name.

Procedure

1. Navigate to the +Add view.
2. Click Event Sink in the Eventing catalog.
3. Search for KafkaSink in the catalog items and click it.
4. Click Create Event Sink.

5. In the form view, type the URL of the bootstrap server, which is a combination of host name and port.

   

   View larger image

   
6. Type the name of the topic to send event data.
7. Type the name of the event sink.
8. Click Create.

Verification

1. Navigate to the Topology view.
2. Click the created event sink to view its details in the right panel.

Procedure

1. Create the certificate files as a secret in the same namespace as your KafkaSink object:

   Important

   Certificates and keys must be in PEM format.

   • For authentication using SASL without encryption:

     $ oc create secret -n <namespace> generic <secret_name> \
       --from-literal=protocol=SASL_PLAINTEXT \
       --from-literal=sasl.mechanism=<sasl_mechanism> \
       --from-literal=user=<username> \
       --from-literal=password=<password>

     Copy to Clipboard Toggle word wrap

   • For authentication using SASL and encryption using TLS:

     $ oc create secret -n <namespace> generic <secret_name> \
       --from-literal=protocol=SASL_SSL \
       --from-literal=sasl.mechanism=<sasl_mechanism> \
       --from-file=ca.crt=<my_caroot.pem_file_path> \ 

     1

     
       --from-literal=user=<username> \
       --from-literal=password=<password>

     Copy to Clipboard Toggle word wrap

     1

     The ca.crt can be omitted to use the system’s root CA set if you are using a public cloud managed Kafka service.

   • For authentication and encryption using TLS:

     $ oc create secret -n <namespace> generic <secret_name> \
       --from-literal=protocol=SSL \
       --from-file=ca.crt=<my_caroot.pem_file_path> \ 

     1

     
       --from-file=user.crt=<my_cert.pem_file_path> \
       --from-file=user.key=<my_key.pem_file_path>

     Copy to Clipboard Toggle word wrap

     1

     The ca.crt can be omitted to use the system’s root CA set if you are using a public cloud managed Kafka service.

2. Create or modify a KafkaSink object and add a reference to your secret in the auth spec:

   apiVersion: eventing.knative.dev/v1alpha1
   kind: KafkaSink
   metadata:
      name: <sink_name>
      namespace: <namespace>
   spec:
   ...
      auth:
        secret:
          ref:
            name: <secret_name>
   ...

   Copy to Clipboard Toggle word wrap

3. Apply the KafkaSink object:

   $ oc apply -f <filename>

   Copy to Clipboard Toggle word wrap

Procedure

1. Create a JobSink object definition as a YAML file:

   JobSink YAML

   apiVersion: sinks.knative.dev/v1alpha1
   kind: JobSink
   metadata:
     name: job-sink-logger
   spec:
     job:
       spec:
         completions: 1
         parallelism: 1
         template:
           spec:
             restartPolicy: Never
             containers:
               - name: main
                 image: docker.io/library/bash:5
                 command: [ "cat" ]
                 args:
                   - "/etc/jobsink-event/event"

   Copy to Clipboard Toggle word wrap

2. Apply the JobSink YAML file:

   $ oc apply -f <job-sink-file.yaml>

   Copy to Clipboard Toggle word wrap

3. Verify JobSink is ready:

   $ oc get jobsinks.sinks.knative.dev

   Copy to Clipboard Toggle word wrap

   Example output:

   NAME              URL                                                                          AGE   READY   REASON
   job-sink-logger   http://job-sink.knative-eventing.svc.cluster.local/default/job-sink-logger   5s    True

   Copy to Clipboard Toggle word wrap

4. Trigger a JobSink. JobSink can be triggered by any event source or trigger.

   $ oc run curl --image=curlimages/curl --rm=true --restart=Never -ti -- -X POST -v \
      -H "content-type: application/json"  \
      -H "ce-specversion: 1.0" \
      -H "ce-source: my/curl/command" \
      -H "ce-type: my.demo.event" \
      -H "ce-id: 123" \
      -d '{"details":"JobSinkDemo"}' \
      http://job-sink.knative-eventing.svc.cluster.local/default/job-sink-logger

   Copy to Clipboard Toggle word wrap

5. Verify a Job is created:

   $ oc logs job-sink-loggerszoi6-dqbtq

   Copy to Clipboard Toggle word wrap

   Example output:

   {"specversion":"1.0","id":"123","source":"my/curl/command","type":"my.demo.event","datacontenttype":"application/json","data":{"details":"JobSinkDemo"}}

   Copy to Clipboard Toggle word wrap