Using the Streams for Apache Kafka Bridge

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>/application.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.

Procedure

1. Edit the application.properties file provided with the 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.

      For example:

      bridge.id=my-bridge
      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.

2. Save the configuration file.

Procedure

1. Set the bridge.metrics configuration to jmxPrometheusExporter.

   Configuration for enabling metrics

   bridge.metrics=jmxPrometheusExporter

   Copy to Clipboard Toggle word wrap

   Optionally, you can add a custom Prometheus JMX Exporter configuration using the bridge.metrics.exporter.config.path property. If not configured, a default embedded configuration file is used.

2. Run the Kafka Bridge run script.

   Running the Kafka Bridge

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

   Copy to Clipboard Toggle word wrap

   With metrics enabled, you can scrape metrics in Prometheus format from the /metrics endpoint of the Kafka Bridge.

Procedure

1. Set the bridge.metrics configuration to strimziMetricsReporter.

   Configuration for enabling metrics

   bridge.metrics=strimziMetricsReporter

   Copy to Clipboard Toggle word wrap

   Optionally, you can configure a comma-separated list of regular expressions to filter exposed metrics using the kafka.prometheus.metrics.reporter.allowlist property. If not configured, a default set of metrics is exposed.

   When needed, it is possible to configure the allowlist per client type. For example, by using the kafka.admin prefix and setting kafka.admin.prometheus.metrics.reporter.allowlist=, all admin client metrics are excluded.

   You can add any plugin configuration to the Kafka Bridge properties file using kafka., kafka.admin., kafka.producer., and kafka.consumer. prefixes. In the event that the same property is configured with multiple prefixes, the most specific prefix takes precedence. For example, kafka.producer.prometheus.metrics.reporter.allowlist takes precedence over kafka.prometheus.metrics.reporter.allowlist.

2. Run the Kafka Bridge run script.

   Running the Kafka Bridge

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

   Copy to Clipboard Toggle word wrap

   With metrics enabled, you can scrape metrics in Prometheus format from the /metrics endpoint of the Kafka Bridge.

Procedure

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

   Use the bridge.tracing property to enable the tracing you want to use.

   Example configuration to enable OpenTelemetry

   bridge.tracing=opentelemetry 

   1

   Copy to Clipboard Toggle word wrap

   1

   The property for enabling OpenTelemetry is uncommented by removing the # at the beginning of the line.

   With tracing enabled, you initialize tracing when you run the Kafka Bridge script.

2. Save the configuration file.

3. Set the environment variables for tracing.

   Environment variables for OpenTelemetry

   OTEL_SERVICE_NAME=my-tracing-service 

   1

   
   OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 

   2

   Copy to Clipboard Toggle word wrap

   1

   The name of the OpenTelemetry tracer service.

   2

   The gRPC-based OTLP endpoint that listens for spans on port 4317.

4. Run the Kafka Bridge script with the property enabled for tracing.

   Running the Kafka Bridge with OpenTelemetry enabled

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

   Copy to Clipboard Toggle word wrap

   The internal consumers and producers of the Kafka Bridge are now enabled for tracing.