Knatve Sink and SinkBinding

In this section we will connect the Knative Services (refer to previous section) to Kafka using Knative Sink and SinkBinding.

SinkBinding supports decoupling the source (service which produces events) from the actual sink by injecting the sink URL into the services.

The red box in the diagram below highlights the Knative Sinks and Sink binding that you will be creating within this section.

1. Create Sink and SinkBinding

This solution needs a number of Sinks and SinkBinding for the various Kafka topics described in an earlier section. You will create one of them here, while the others have been pre-configured for you.

Here is a visual of how the reviews flows from the User to Kafka with Knative eventing.

The reviews submitted by the user are sent to the product_reviews Quarkus service through HTTP POST.
The product_reviews service sends this review as a CloudEvent to the reviews-sink Kafka Sink over HTTP.
The Quarkus service remains agnostic to the internals of the Kafka streaming platform.
The reviews-sink Kafka Sink sends this Cloud Event to the globex.reviews Kafka topic.

Now, go ahead and create the Sink and SinkBinding.

Navigate to OpenShift console {openshift_cluster_console}[OpenShift Console^, window="console"] and if needed login with ({user_name}/{user_password}).
From the the top right corner of the OpenShift console, click on the (+) button and then Import Yaml dropdown option

Copy the following YAML into the Import YAML form, and click Create to create the KafkaSink reviews-sink which will send messages to globex.reviews Kafka Topic.

apiVersion: eventing.knative.dev/v1alpha1
kind: KafkaSink
metadata:
  name: reviews-sink
  namespace: globex-serverless-{user_name}
spec:
  bootstrapServers:
    - kafka-kafka-bootstrap.globex-mw-{user_name}.svc.cluster.local:9092
  topic: globex.reviews
  numPartitions: 1
  contentMode: binary
  auth:
     secret:
       ref:
         name: kafka-secret

Use the Import YAML form to create a Sink Binding from the product-reviews Quarkus Service to the KafkaSink reviews-sink that you created in the previous step.

apiVersion: sources.knative.dev/v1
kind: SinkBinding
metadata:
  name: product-reviews-to-reviews-sink
  namespace: globex-serverless-{user_name}
spec:
  sink:
    ref:
      apiVersion: eventing.knative.dev/v1alpha1
      kind: KafkaSink
      name: reviews-sink
      namespace: globex-serverless-{user_name}
  subject:
    apiVersion: apps/v1
    kind: Deployment
    name: product-reviews
    namespace: globex-serverless-{user_name}

Navigate back to the {openshift_cluster_console}/topology/ns/globex-serverless-{user_name}?view=graph[Topology View, window="console", target="console"], to view the new Sink and SinkBinding you created

Here is the list of all the Kafka Sinks used in this solution.

Sink name

Function

reviews-sink

Send the reviews submitted by user (HTTP POST from globex-web app to product-reviews Quarkus service) as CloudEvents to globex.reviews Kafka topic

moderated-reviews-sink

Sends reviews moderated by the aiml-moderate-reviews service to topic reviews.moderated

reviews-sentiment-sink

Sends sentiment score of reviews by the aiml-sentiment-reviews service to topic reviews.sentiment

2. A note on KafkaSink and SinkBinding objects

A Kafka Sink persists incoming CloudEvents to a configurable Apache Kafka Topic.
A SinkBinding object supports decoupling event production (the producer of the CloudEvent) from delivery addressing (knowing the actual Kafka Sink’s URL).

In this case

The reviews submitted by the user is will be sent by the product-reviews service to the reviews-sink KafkaSink via HTTP Post as a CloudEvent.
This KafkaSink persists this CloudEvent in the globex.reviews Kafka topic.

The product-reviews-to-reviews-sink SinkBinding that you created updated the product-reviews service’s environment variable K_SINK with the the Kafka Sink’s URL on runtime. Because of this, the application code does not need to interact directly with the Kubernetes API to locate the event destination.

{openshift_cluster_console}/k8s/ns/globex-serverless-{user-name}/deployments/product-reviews/environment[Click to view^] the environment variables in the product-reviews service’s deployment configuration. You will see that the K_SINK value matches the status.address.url of the reviews-sink {openshift_cluster_console}/k8s/ns/globex-serverless-{user-name}/eventing.knative.dev%7Ev1alpha1%7EKafkaSink/reviews-sink/yaml[Kafka Sink CR^].