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:

   The following example displays 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

   • name: <source_name>: The name of the event source.
   • namespace: <namespace>: The namespace of the event source.
   • serviceAccountName: <service_account_name>: The service account for the event source.
   • name: <sink_name>: 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 hostname 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. 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 use Privacy-Enhanced Mail (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> \
       --from-literal=user=<username> \
       --from-literal=password=<password>

     from-file=ca.crt=<my_caroot.pem_file_path>: 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> \
       --from-file=user.crt=<my_cert.pem_file_path> \
       --from-file=user.key=<my_key.pem_file_path>

     from-file=ca.crt=<my_caroot.pem_file_path>: 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 by running the following command:

   $ oc apply -f <filename>

Procedure

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

   The following example displays 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 by running the following command:

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

3. Verify JobSink is ready by running the following command:

   $ oc get jobsinks.sinks.knative.dev

   You get an output similar to the following example:

   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 by running the following command:

   $ oc logs job-sink-loggerszoi6-dqbtq

   You get an output similar to the following example:

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