REST Proxy: Example for Apache Kafka®¶
In this tutorial, you will use Confluent REST Proxy to produce messages to and consumes messages from an Apache Kafka® cluster.
After you run the tutorial, use the provided source code as a reference to develop your own Kafka client application.
Prerequisites¶
Client¶
- Docker version 17.06.1-ce
- Docker Compose version 1.25.4
wget
Kafka Cluster¶
The easiest way to follow this tutorial is with Confluent Cloud because you don’t have to run a local Kafka cluster.
When you sign up for Confluent Cloud, apply promo code C50INTEG
to receive an additional $50 free usage (details).
From the Console, click on LEARN to provision a cluster and click on Clients
to get the cluster-specific configurations and credentials to set for your client application.
You can alternatively use the supported CLI or REST API, or the community-supported ccloud-stack Utility for Confluent Cloud.
If you don’t want to use Confluent Cloud, you can also use this tutorial with a Kafka cluster running on your local host or any other remote server.
Setup¶
Clone the confluentinc/examples GitHub repository and check out the
6.2.0-post
branch.git clone https://github.com/confluentinc/examples cd examples git checkout 6.2.0-post
Change directory to the example for REST Proxy.
cd clients/cloud/rest-proxy/
Create a local file (for example, at
$HOME/.confluent/java.config
) with configuration parameters to connect to your Kafka cluster. Starting with one of the templates below, customize the file with connection information to your cluster. Substitute your values for{{ BROKER_ENDPOINT }}
,{{CLUSTER_API_KEY }}
, and{{ CLUSTER_API_SECRET }}
(see Configure Confluent Cloud Clients for instructions on how to manually find these values, or use the ccloud-stack Utility for Confluent Cloud to automatically create them).Template configuration file for Confluent Cloud
# Required connection configs for Kafka producer, consumer, and admin bootstrap.servers={{ BROKER_ENDPOINT }} security.protocol=SASL_SSL sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='{{ CLUSTER_API_KEY }}' password='{{ CLUSTER_API_SECRET }}'; sasl.mechanism=PLAIN # Required for correctness in Apache Kafka clients prior to 2.6 client.dns.lookup=use_all_dns_ips # Best practice for Kafka producer to prevent data loss acks=all
Template configuration file for local host
# Kafka bootstrap.servers=localhost:9092
Generate a file of ENV variables used by Docker to set the bootstrap servers and security configuration.
../../../ccloud/ccloud-generate-cp-configs.sh $HOME/.confluent/java.config
Source the generated file of
ENV
variables.source ./delta_configs/env.delta
Get the cp-all-in-one-cloud docker-compose.yml file, which runs Confluent Platform in containers in your local host, and automatically configures them to connect to Confluent Cloud.
wget -O docker-compose.yml https://raw.githubusercontent.com/confluentinc/cp-all-in-one/6.2.0-post/cp-all-in-one-cloud/docker-compose.yml
For the full REST Proxy configuration, view the REST Proxy section in the
docker-compose.yml
file which you just downloaded in the previous step.cat docker-compose.yml
Basic Producer and Consumer¶
In this example, the producer application writes Kafka data to a topic in your Kafka cluster.
If the topic does not already exist in your Kafka cluster, the producer application will use the Kafka Admin Client API to create the topic.
Each record written to Kafka has a key representing a username (for example, alice
) and a value of a count, formatted as json (for example, {"count": 0}
).
The consumer application reads the same Kafka topic and keeps a rolling sum of the count as it processes each record.
Produce Records¶
Since you are not going to use Schema Registry in this section, comment out the following lines in the
docker-compose.yml
file:#KAFKA_REST_SCHEMA_REGISTRY_URL: $SCHEMA_REGISTRY_URL #KAFKA_REST_CLIENT_BASIC_AUTH_CREDENTIALS_SOURCE: $BASIC_AUTH_CREDENTIALS_SOURCE #KAFKA_REST_CLIENT_SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO: $SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO
Start the REST Proxy Docker container by running the following command:
docker-compose up -d rest-proxy
View the REST Proxy logs and wait till you see the log message
Server started, listening for requests
to confirm it has started.docker-compose logs -f rest-proxy
Get the Kafka cluster ID that the REST Proxy is connected to.
KAFKA_CLUSTER_ID=$(docker-compose exec rest-proxy curl -X GET \ "http://localhost:8082/v3/clusters/" | jq -r ".data[0].cluster_id")
Verify the parameter
KAFKA_CLUSTER_ID
has a valid value. For the example in this tutorial, it is shown aslkc-56ngz
, but it will differ in your output.echo $KAFKA_CLUSTER_ID
Create the Kafka topic
test1
using theAdminClient
functionality of the REST Proxy API v3.docker-compose exec rest-proxy curl -X POST \ -H "Content-Type: application/json" \ -d "{\"topic_name\":\"test1\",\"partitions_count\":6,\"configs\":[]}" \ "http://localhost:8082/v3/clusters/${KAFKA_CLUSTER_ID}/topics" | jq .
Verify your output resembles:
{ "kind": "KafkaTopic", "metadata": { "self": "http://rest-proxy:8082/v3/clusters/lkc-56ngz/topics/test1", "resource_name": "crn:///kafka=lkc-56ngz/topic=test1" }, "cluster_id": "lkc-56ngz", "topic_name": "test2", "is_internal": false, "replication_factor": 0, "partitions": { "related": "http://rest-proxy:8082/v3/clusters/lkc-56ngz/topics/test2/partitions" }, "configs": { "related": "http://rest-proxy:8082/v3/clusters/lkc-56ngz/topics/test2/configs" }, "partition_reassignments": { "related": "http://rest-proxy:8082/v3/clusters/lkc-56ngz/topics/test1/partitions/-/reassignment" } }
Produce three JSON messages to the topic, with key
alice
, and values{"count":0}
,{"count":1}
, and{"count":2}
.docker-compose exec rest-proxy curl -X POST \ -H "Content-Type: application/vnd.kafka.json.v2+json" \ -H "Accept: application/vnd.kafka.v2+json" \ --data '{"records":[{"key":"alice","value":{"count":0}},{"key":"alice","value":{"count":1}},{"key":"alice","value":{"count":2}}]}' \ "http://localhost:8082/topics/test1" | jq .
Verify your output resembles:
{ "offsets": [ { "partition": 0, "offset": 0, "error_code": null, "error": null }, { "partition": 0, "offset": 1, "error_code": null, "error": null }, { "partition": 0, "offset": 2, "error_code": null, "error": null } ], "key_schema_id": null, "value_schema_id": null }
View the producer code.
Consume Records¶
Create a consumer
ci1
belonging to consumer groupcg1
. Specifyauto.offset.reset
to beearliest
so it starts at the beginning of the topic.docker-compose exec rest-proxy curl -X POST \ -H "Content-Type: application/vnd.kafka.v2+json" \ --data '{"name": "ci1", "format": "json", "auto.offset.reset": "earliest"}' \ http://localhost:8082/consumers/cg1 | jq .
Verify your output resembles:
{ "instance_id": "ci1", "base_uri": "http://rest-proxy:8082/consumers/cg1/instances/ci1" }
Subscribe the consumer to topic
test1
.docker-compose exec rest-proxy curl -X POST \ -H "Content-Type: application/vnd.kafka.v2+json" \ --data '{"topics":["test1"]}' \ http://localhost:8082/consumers/cg1/instances/ci1/subscription | jq .
Consume data using the base URL in the first response. Issue the curl command twice, sleeping 10 seconds in between—this is intentional due to https://github.com/confluentinc/kafka-rest/issues/432.
docker-compose exec rest-proxy curl -X GET \ -H "Accept: application/vnd.kafka.json.v2+json" \ http://localhost:8082/consumers/cg1/instances/ci1/records | jq . sleep 10 docker-compose exec rest-proxy curl -X GET \ -H "Accept: application/vnd.kafka.json.v2+json" \ http://localhost:8082/consumers/cg1/instances/ci1/records | jq .
Verify your output resembles:
[] [ { "topic": "test1", "key": "alice", "value": { "count": 0 }, "partition": 0, "offset": 0 }, { "topic": "test1", "key": "alice", "value": { "count": 1 }, "partition": 0, "offset": 1 }, { "topic": "test1", "key": "alice", "value": { "count": 2 }, "partition": 0, "offset": 2 } ]
Delete the consumer instance to clean up its resources
docker-compose exec rest-proxy curl -X DELETE \ -H "Content-Type: application/vnd.kafka.v2+json" \ http://localhost:8082/consumers/cg1/instances/ci1 | jq .
View the consumer code.
Stop REST Proxy¶
Stop Docker by running the following command:
docker-compose down
Avro and Confluent Cloud Schema Registry¶
This example is similar to the previous example, except the value is formatted as Avro and integrates with the Confluent Cloud Schema Registry. Before using Confluent Cloud Schema Registry, check its availability and limits.
As described in the Quick Start for Schema Management on Confluent Cloud in the Confluent Cloud Console, enable Confluent Cloud Schema Registry and create an API key and secret to connect to it.
Verify that your VPC can connect to the Confluent Cloud Schema Registry public internet endpoint.
Update your local configuration file (for example, at
$HOME/.confluent/java.config
) with parameters to connect to Schema Registry.Template configuration file for Confluent Cloud
# Required connection configs for Kafka producer, consumer, and admin bootstrap.servers={{ BROKER_ENDPOINT }} security.protocol=SASL_SSL sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='{{ CLUSTER_API_KEY }}' password='{{ CLUSTER_API_SECRET }}'; sasl.mechanism=PLAIN # Required for correctness in Apache Kafka clients prior to 2.6 client.dns.lookup=use_all_dns_ips # Best practice for Kafka producer to prevent data loss acks=all # Required connection configs for Confluent Cloud Schema Registry schema.registry.url=https://{{ SR_ENDPOINT }} basic.auth.credentials.source=USER_INFO basic.auth.user.info={{ SR_API_KEY }}:{{ SR_API_SECRET }}
Template configuration file for local host
# Kafka bootstrap.servers=localhost:9092 # Confluent Schema Registry schema.registry.url=http://localhost:8081
Verify your Confluent Cloud Schema Registry credentials by listing the Schema Registry subjects. In the following example, substitute your values for
{{ SR_API_KEY }}
,{{ SR_API_SECRET }}
, and{{ SR_ENDPOINT }}
.curl -u {{ SR_API_KEY }}:{{ SR_API_SECRET }} https://{{ SR_ENDPOINT }}/subjects
Regenerate a file of ENV variables used by Docker to set the bootstrap servers and security configuration.
../../../ccloud/ccloud-generate-cp-configs.sh $HOME/.confluent/java.config
Source the regenerated file of
ENV
variables.source ./delta_configs/env.delta
Produce Avro Records¶
Since you are now going to use Schema Registry in this section, uncomment the following lines in the
docker-compose.yml
file:KAFKA_REST_SCHEMA_REGISTRY_URL: $SCHEMA_REGISTRY_URL KAFKA_REST_CLIENT_BASIC_AUTH_CREDENTIALS_SOURCE: $BASIC_AUTH_CREDENTIALS_SOURCE KAFKA_REST_CLIENT_SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO: $SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO
Start the REST Proxy Docker container by running the following command:
docker-compose up -d rest-proxy
View the REST Proxy logs and wait till you see the log message
Server started, listening for requests
to confirm it has started.docker-compose logs -f rest-proxy
Get the Kafka cluster ID that the REST Proxy is connected to.
KAFKA_CLUSTER_ID=$(docker-compose exec rest-proxy curl -X GET \ "http://localhost:8082/v3/clusters/" | jq -r ".data[0].cluster_id")
Verify the parameter
KAFKA_CLUSTER_ID
has a valid value. For the example in this tutorial, it is shown aslkc-56ngz
, but it will differ in your output.Create the Kafka topic
test2
using theAdminClient
functionality of the REST Proxy API v3.docker-compose exec rest-proxy curl -X POST \ -H "Content-Type: application/json" \ -d "{\"topic_name\":\"test2\",\"partitions_count\":6,\"configs\":[]}" \ "http://localhost:8082/v3/clusters/${KAFKA_CLUSTER_ID}/topics" | jq .
Verify your output resembles:
{ "kind": "KafkaTopic", "metadata": { "self": "http://rest-proxy:8082/v3/clusters/lkc-56ngz/topics/test2", "resource_name": "crn:///kafka=lkc-56ngz/topic=test2" }, "cluster_id": "lkc-56ngz", "topic_name": "test2", "is_internal": false, "replication_factor": 0, "partitions": { "related": "http://rest-proxy:8082/v3/clusters/lkc-56ngz/topics/test2/partitions" }, "configs": { "related": "http://rest-proxy:8082/v3/clusters/lkc-56ngz/topics/test2/configs" }, "partition_reassignments": { "related": "http://rest-proxy:8082/v3/clusters/lkc-56ngz/topics/test2/partitions/-/reassignment" } }
Register a new Avro schema for topic
test2
with the Confluent Cloud Schema Registry.docker-compose exec rest-proxy curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{ "schema": "[ { \"type\":\"record\", \"name\":\"countInfo\", \"fields\": [ {\"name\":\"count\",\"type\":\"long\"}]} ]" }' -u "$SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO" "$SCHEMA_REGISTRY_URL/subjects/test2-value/versions"
Verify the output shows the new schema id:
{"id":100001}
Set the variable
schemaid
to the value of the schema ID.schemaid=$(docker-compose exec rest-proxy curl -X GET -u "$SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO" "$SCHEMA_REGISTRY_URL/subjects/test2-value/versions/latest" | jq '.id')
Produce three Avro messages to the topic, with values
{"count":0}
,{"count":1}
, and{"count":2}
. Notice that the request body includes the schema ID.docker-compose exec rest-proxy curl -X POST \ -H "Content-Type: application/vnd.kafka.avro.v2+json" \ -H "Accept: application/vnd.kafka.v2+json" \ --data '{"value_schema_id": '"$schemaid"', "records": [{"value": {"countInfo":{"count": 0}}},{"value": {"countInfo":{"count": 1}}},{"value": {"countInfo":{"count": 2}}}]}' \ "http://localhost:8082/topics/test2" | jq .
Verify your output resembles:
{ "offsets": [ { "partition": 4, "offset": 0, "error_code": null, "error": null }, { "partition": 4, "offset": 1, "error_code": null, "error": null }, { "partition": 4, "offset": 2, "error_code": null, "error": null } ], "key_schema_id": null, "value_schema_id": 100001 }
View the producer Avro code.
Consume Avro Records¶
Create a consumer
ci2
belonging to consumer groupcg2
. Specifyauto.offset.reset
to beearliest
so it starts at the beginning of the topic.docker-compose exec rest-proxy curl -X POST \ -H "Content-Type: application/vnd.kafka.v2+json" \ --data '{"name": "ci2", "format": "avro", "auto.offset.reset": "earliest"}' \ http://localhost:8082/consumers/cg2 | jq .
Verify your output resembles:
{ "instance_id": "ci2", "base_uri": "http://rest-proxy:8082/consumers/cg2/instances/ci2" }
Subscribe the consumer to topic
test2
.docker-compose exec rest-proxy curl -X POST \ -H "Content-Type: application/vnd.kafka.v2+json" \ --data '{"topics":["test2"]}' \ http://localhost:8082/consumers/cg2/instances/ci2/subscription | jq .
Consume data using the base URL in the first response. Issue the curl command twice, sleeping 10 seconds in between—this is intentional due to https://github.com/confluentinc/kafka-rest/issues/432.
docker-compose exec rest-proxy curl -X GET \ -H "Accept: application/vnd.kafka.avro.v2+json" \ http://localhost:8082/consumers/cg2/instances/ci2/records | jq . sleep 10 docker-compose exec rest-proxy curl -X GET \ -H "Accept: application/vnd.kafka.avro.v2+json" \ http://localhost:8082/consumers/cg2/instances/ci2/records | jq .
Verify your output resembles:
[] [ { "topic": "test2", "key": null, "value": { "count": 0 }, "partition": 0, "offset": 0 }, { "topic": "test2", "key": null, "value": { "count": 1 }, "partition": 0, "offset": 1 }, { "topic": "test2", "key": null, "value": { "count": 2 }, "partition": 0, "offset": 2 } ]
Delete the consumer instance to clean up its resources
docker-compose exec rest-proxy curl -X DELETE \ -H "Content-Type: application/vnd.kafka.v2+json" \ http://localhost:8082/consumers/cg2/instances/ci2 | jq .
View the consumer Avro code.
Confluent Cloud Schema Registry¶
View the schema subjects registered in Confluent Cloud Schema Registry. In the following output, substitute values for
<SR API KEY>
,<SR API SECRET>
, and<SR ENDPOINT>
.curl -u <SR API KEY>:<SR API SECRET> https://<SR ENDPOINT>/subjects
Verify that the subject
test2-value
exists.["test2-value"]
View the schema information for subject test2-value. In the following output, substitute values for
<SR API KEY>
,<SR API SECRET>
, and<SR ENDPOINT>
.curl -u <SR API KEY>:<SR API SECRET> https://<SR ENDPOINT>/subjects/test2-value/versions/1
Verify the schema information for subject
test2-value
.{"subject":"test2-value","version":1,"id":100001,"schema":"[{\"type\":\"record\",\"name\":\"countInfo\",\"fields\":[{\"name\":\"count\",\"type\":\"long\"}]}]"}
Stop REST Proxy¶
Stop Docker by running the following command:
docker-compose down