Chapter 11. Kafka Bridge

Procedure

1. Edit the application.properties file provided with the AMQ Streams Kafka Bridge installation archive.

   Use the properties file to specify Kafka and HTTP-related properties.

   1. Configure standard Kafka-related properties, including properties specific to the Kafka consumers and producers.

      Use:

      • kafka.bootstrap.servers to define the host/port connections to the Kafka cluster
      • kafka.producer.acks to provide acknowledgments to the HTTP client

      • kafka.consumer.auto.offset.reset to determine how to manage reset of the offset in Kafka

        For more information on configuration of Kafka properties, see the Apache Kafka website

   2. Configure HTTP-related properties to enable HTTP access to the Kafka cluster.

      http.enabled=true
      http.host=0.0.0.0
      http.port=8080

Procedure

1. If you have not already done so, unzip the AMQ Streams Kafka Bridge installation archive to any directory.

2. Run the Kafka Bridge script using the configuration properties as a parameter:

   For example:

   ./bin/kafka_bridge_run.sh --config-file=_path_/configfile.properties

3. Check to see that the installation was successful in the log.

   HTTP-Kafka Bridge started and listening on port 8080
   HTTP-Kafka Bridge bootstrap servers localhost:9092

Procedure

1. Open the application.properties file and check that the default HTTP related settings are defined:

   http.enabled=true
   http.host=0.0.0.0
   http.port=8080

   This configures the Kafka Bridge to listen for requests on port 8080.

2. Run the Kafka Bridge script using the configuration properties as a parameter:

   ./bin/kafka_bridge_run.sh --config-file=<path>/application.properties

Procedure

1. Create a Kafka topic using the kafka-topics.sh utility:

   bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic bridge-quickstart-topic --partitions 3 --replication-factor 1 --config retention.ms=7200000 --config segment.bytes=1073741824

   Specify three partitions.

2. Verify that the topic was created:

   bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic bridge-quickstart-topic

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

4. If the request is successful, the Kafka Bridge returns an offsets array, along with a 200 (OK) 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.

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

     Note

     The Kafka Bridge generates a random consumer name if you do not specify a consumer name in the request body.

     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 (OK) 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 triggers 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 (OK) 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 No Content 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 No Content code.