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>

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

   $ oc apply -f <filename>

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

   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. In the Developer perspective, 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.

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

Verification

1. In the Developer perspective, 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>

   • 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>

     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>

     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>
   ...

3. Apply the KafkaSink object:

   $ oc apply -f <filename>

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"

2. Apply the JobSink YAML file:

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

3. Verify JobSink is ready:

   $ oc get jobsinks.sinks.knative.dev

   Example output:

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

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

5. Verify a Job is created:

   $ oc logs job-sink-loggerszoi6-dqbtq

   Example output:

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