Chapter 6. Kafka Bridge

Procedure

1. Edit the examples/bridge/kafka-bridge.yaml file.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaBridge
   metadata:
     name: quickstart 1
   spec:
     replicas: 1
     bootstrapServers: <cluster-name>-kafka-bootstrap:9092 2
     http:
       port: 8080

   1

   When the Kafka Bridge is deployed, -bridge is appended to the name of the deployment and other related resources. In this example, the Kafka Bridge deployment is named quickstart-bridge and the accompanying Kafka Bridge service is named quickstart-bridge-service.

   2

   In the bootstrapServers property, enter the name of the Kafka cluster as the <cluster-name>.

2. Deploy the Kafka Bridge to your OpenShift cluster:

   oc apply -f examples/bridge/kafka-bridge.yaml

   A quickstart-bridge deployment, service, and other related resources are created in your OpenShift cluster.

3. Verify that the Kafka Bridge was successfully deployed:

   oc get deployments

   NAME                             READY   UP-TO-DATE   AVAILABLE   AGE
   quickstart-bridge                  1/1     1            1          34m
   my-cluster-connect                 1/1     1            1          24h
   my-cluster-entity-operator         1/1     1            1          24h
   #...

Procedure

1. List the names of the pods in your OpenShift cluster:

   oc get pods -o name
   
   pod/kafka-consumer
   # ...
   pod/quickstart-bridge-589d78784d-9jcnr
   pod/strimzi-cluster-operator-76bcf9bc76-8dnfm

2. Connect to the quickstart-bridge pod on port 8080:

   oc port-forward pod/quickstart-bridge-589d78784d-9jcnr 8080:8080 &

   Note

   If port 8080 on your local machine is already in use, use an alternative HTTP port, such as 8008.

Procedure

1. In a text editor, create a YAML definition for a Kafka topic with three partitions.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaTopic
   metadata:
     name: bridge-quickstart-topic
     labels:
       strimzi.io/cluster: <kafka-cluster-name> 1
   spec:
     partitions: 3 2
     replicas: 1
     config:
       retention.ms: 7200000
       segment.bytes: 1073741824

   1

   The name of the Kafka cluster in which the Kafka Bridge is deployed.

   2

   The number of partitions for the topic.

2. Save the file to the examples/topic directory as bridge-quickstart-topic.yaml.

3. Create the topic in your OpenShift cluster:

   oc apply -f examples/topic/bridge-quickstart-topic.yaml

4. Using the Kafka Bridge, produce three messages to the topic you created:

   curl -X POST \
     http://localhost:8080/topics/bridge-quickstart-topic \
     -H 'content-type: application/vnd.kafka.json.v2+json' \
     -d '{
       "records": [
           {
               "key": "my-key",
               "value": "sales-lead-0001"
           },
           {
               "value": "sales-lead-0002",
               "partition": 2
           },
           {
               "value": "sales-lead-0003"
           }
       ]
   }'

   • sales-lead-0001 is sent to a partition based on the hash of the key.
   • sales-lead-0002 is sent directly to partition 2.
   • sales-lead-0003 is sent to a partition in the bridge-quickstart-topic topic using a round-robin method.

5. If the request is successful, the Kafka Bridge returns an offsets array, along with a 200 code and a content-type header of application/vnd.kafka.v2+json. For each message, the offsets array describes:

   • The partition that the message was sent to

   • The current message offset of the partition

     Example response

     #...
     {
       "offsets":[
         {
           "partition":0,
           "offset":0
         },
         {
           "partition":2,
           "offset":0
         },
         {
           "partition":0,
           "offset":1
         }
       ]
     }

Procedure

1. Create a Kafka Bridge consumer in a new consumer group named bridge-quickstart-consumer-group:

   curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group \
     -H 'content-type: application/vnd.kafka.v2+json' \
     -d '{
       "name": "bridge-quickstart-consumer",
       "auto.offset.reset": "earliest",
       "format": "json",
       "enable.auto.commit": false,
       "fetch.min.bytes": 512,
       "consumer.request.timeout.ms": 30000
     }'

   • The consumer is named bridge-quickstart-consumer and the embedded data format is set as json.
   • Some basic configuration settings are defined.

   • The consumer will not commit offsets to the log automatically because the enable.auto.commit setting is false. You will commit the offsets manually later in this quickstart.

     If the request is successful, the Kafka Bridge returns the consumer ID (instance_id) and base URL (base_uri) in the response body, along with a 200 code.

     Example response

     #...
     {
       "instance_id": "bridge-quickstart-consumer",
       "base_uri":"http://<bridge-name>-bridge-service:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer"
     }

2. Copy the base URL (base_uri) to use in the other consumer operations in this quickstart.

Procedure

1. Produce additional messages to the Kafka Bridge consumer, as described in Producing messages to topics and partitions.

2. Submit a GET request to the records endpoint:

   curl -X GET http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/records \
     -H 'accept: application/vnd.kafka.json.v2+json'

   After creating and subscribing to a Kafka Bridge consumer, a first GET request will return an empty response because the poll operation starts a rebalancing process to assign partitions.

3. Repeat step two to retrieve messages from the Kafka Bridge consumer.

   The Kafka Bridge returns an array of messages — describing the topic name, key, value, partition, and offset — in the response body, along with a 200 code. Messages are retrieved from the latest offset by default.

   HTTP/1.1 200 OK
   content-type: application/vnd.kafka.json.v2+json
   #...
   [
     {
       "topic":"bridge-quickstart-topic",
       "key":"my-key",
       "value":"sales-lead-0001",
       "partition":0,
       "offset":0
     },
     {
       "topic":"bridge-quickstart-topic",
       "key":null,
       "value":"sales-lead-0003",
       "partition":0,
       "offset":1
     },
   #...

   Note

   If an empty response is returned, produce more records to the consumer as described in Producing messages to topics and partitions, and then try retrieving messages again.

Procedure

1. Seek to a specific offset for partition 0 of the quickstart-bridge-topic topic:

   curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/positions \
     -H 'content-type: application/vnd.kafka.v2+json' \
     -d '{
       "offsets": [
           {
               "topic": "bridge-quickstart-topic",
               "partition": 0,
               "offset": 2
           }
       ]
   }'

   If the request is successful, the Kafka Bridge returns a 204 code only.

2. Submit a GET request to the records endpoint:

   curl -X GET http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/records \
     -H 'accept: application/vnd.kafka.json.v2+json'

   The Kafka Bridge returns messages from the offset that you seeked to.

3. Restore the default message retrieval behavior by seeking to the last offset for the same partition. This time, use the positions/end endpoint.

   curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/positions/end \
     -H 'content-type: application/vnd.kafka.v2+json' \
     -d '{
       "partitions": [
           {
               "topic": "bridge-quickstart-topic",
               "partition": 0
           }
       ]
   }'

   If the request is successful, the Kafka Bridge returns another 204 code.