REST Proxy¶
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, view the provided source code and use it 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¶
- You can use this tutorial with a Kafka cluster in any environment:
- In Confluent Cloud
- On your local host
- Any remote Kafka cluster
- If you are running on Confluent Cloud, you must have access to a
Confluent Cloud cluster
with an API key and secret.
- The first 20 users to sign up for Confluent Cloud and use promo code
C50INTEG
will receive an additional $50 free usage (details) - For an automated way to create a Kafka cluster, credentials, and ACLs in Confluent Cloud, see ccloud-stack Utility for Confluent Cloud.
- The first 20 users to sign up for Confluent Cloud and use promo code
Setup¶
Clone the confluentinc/examples GitHub repository and check out the
6.0.0-post
branch.git clone https://github.com/confluentinc/examples cd examples git checkout 6.0.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.0.0/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
Start the REST Proxy Docker container by running the following command:
docker-compose up -d rest-proxy
View the REST Proxy logs in Docker and wait till you see the log message
Server started, listening for requests
to confirm REST Proxy has started.docker-compose logs -f rest-proxy
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¶
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. If REST Proxy is backed to Confluent Cloud, configure the replication factor to3
.docker-compose exec rest-proxy curl -X POST \ -H "Content-Type: application/json" \ -d "{\"topic_name\":\"test1\",\"partitions_count\":6,\"replication_factor\":3,\"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": 3, "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.
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 GUI, 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 schema.registry.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 work from your host. In the following example, substitute your values for
{{ SR_API_KEY}}
,{{SR_API_SECRET }}
, and{{ SR_ENDPOINT }}
.# View the list of registered subjects $ curl -u {{ SR_API_KEY }}:{{ SR_API_SECRET }} https://{{ SR_ENDPOINT }}/subjects # Same as above, as a single bash command to parse the values out of $HOME/.confluent/java.config $ curl -u $(grep "^schema.registry.basic.auth.user.info" $HOME/.confluent/java.config | cut -d'=' -f2) $(grep "^schema.registry.url" $HOME/.confluent/java.config | cut -d'=' -f2)/subjects
Produce Avro Records¶
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. If REST Proxy is backed to Confluent Cloud, configure the replication factor to3
.docker-compose exec rest-proxy curl -X POST \ -H "Content-Type: application/json" \ -d "{\"topic_name\":\"test2\",\"partitions_count\":6,\"replication_factor\":3,\"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": 3, "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¶
Stop Docker by running the following command:
docker-compose down