Using the AMQ Streams 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, and to enable distributed tracing.

   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.

      For example:

      bridge.id=my-bridge
      http.enabled=true
      http.host=0.0.0.0
      http.port=8080 

      1

      
      http.cors.enabled=true 

      2

      
      http.cors.allowedOrigins=https://strimzi.io 

      3

      
      http.cors.allowedMethods=GET,POST,PUT,DELETE,OPTIONS,PATCH 

      4

      Copy to Clipboard Toggle word wrap

      1

      The default HTTP configuration for the Kafka Bridge to listen on port 8080.

      2

      Set to true to enable CORS.

      3

      Comma-separated list of allowed CORS origins. You can use a URL or a Java regular expression.

      4

      Comma-separated list of allowed HTTP methods for CORS.

Procedure

1. If you have not already done so, unzip the 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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

Procedure

1. 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"
           }
       ]
   }'

   Copy to Clipboard Toggle word wrap
   • 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.

2. 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
         }
       ]
     }

     Copy to Clipboard Toggle word wrap

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
     }'

   Copy to Clipboard Toggle word wrap
   • 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_id>-bridge-service:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer"
     }

     Copy to Clipboard Toggle word wrap

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'

   Copy to Clipboard Toggle word wrap

   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
     },
   #...

   Copy to Clipboard Toggle word wrap
   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
           }
       ]
   }'

   Copy to Clipboard Toggle word wrap

   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'

   Copy to Clipboard Toggle word wrap

   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
           }
       ]
   }'

   Copy to Clipboard Toggle word wrap

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