<commit_before>.. _kafkarest_intro:

Kafka REST Proxy
================

The Kafka REST Proxy provides a RESTful interface to a Kafka cluster. It makes
it easy to produce and consume messages, view the state of the cluster, and
perform administrative actions without using the native Kafka protocol or
clients. Examples of use cases include reporting data to Kafka from any
frontend app built in any language, ingesting messages into a stream processing
framework that doesn't yet support Kafka, and scripting administrative actions.

Quickstart
----------

Start by running the REST Proxy and the services it depends on: ZooKeeper, Kafka, and the Schema
Registry:

.. sourcecode:: bash

   $ ./bin/zookeeper-server-start ./etc/kafka/zookeeper.properties &
   $ ./bin/kafka-server-start ./etc/kafka/server.properties &
   $ ./bin/schema-registry-start ./etc/schema-registry/schema-registry.properties &
   $ ./bin/kafka-rest-start ./etc/kafka-rest/kafka-rest.properties &

.. ifconfig:: platform_docs

   See the :ref:`Confluent Platform quickstart<quickstart>` for a more detailed explanation of how
   to get these services up and running.

Inspect Topic Metadata
~~~~~~~~~~~~~~~~~~~~~~

.. sourcecode:: bash

   # Get a list of topics
   $ curl "http://localhost:8082/topics"
     ["test","test2","test3"]

   # Get info about one topic
   $ curl "http://localhost:8082/topics/test"
     {"test":"connect-test","configs":{},"partitions":[{"partition":0,"leader":0,"replicas":[{"broker":0,"leader":true,"in_sync":true}]},{"partition":1,"leader":0,"replicas":[{"broker":1,"leader":true,"in_sync":true}]}]}

   # Get info about a topic's partitions
   $ curl "http://localhost:8082/topics/test/partitions"
     [{"partition":0,"leader":0,"replicas":[{"broker":0,"leader":true,"in_sync":true}]},{"partition":1,"leader":0,"replicas":[{"broker":1,"leader":true,"in_sync":true}]}]

Produce and Consume Avro Messages
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. sourcecode:: bash

   # Produce a message using Avro embedded data, including the schema which will
   # be registered with the schema registry and used to validate and serialize
   # before storing the data in Kafka
   $ curl -X POST -H "Content-Type: application/vnd.kafka.avro.v1+json" \
         --data '{"value_schema": "{\"type\": \"record\", \"name\": \"User\", \"fields\": [{\"name\": \"name\", \"type\": \"string\"}]}", "records": [{"value": {"name": "testUser"}}]}' \
         "http://localhost:8082/topics/avrotest"
     {"offsets":[{"partition":0,"offset":0,"error_code":null,"error":null}],"key_schema_id":null,"value_schema_id":21}

   # Create a consumer for Avro data, starting at the beginning of the topic's
   # log. Then consume some data from a topic, which is decoded, translated to
   # JSON, and included in the response. The schema used for deserialization is
   # fetched automatically from the schema registry. Finally, clean up.
   $ curl -X POST -H "Content-Type: application/vnd.kafka.v1+json" \
         --data '{"name": "my_consumer_instance", "format": "avro", "auto.offset.reset": "smallest"}' \
         http://localhost:8082/consumers/my_avro_consumer
     {"instance_id":"my_consumer_instance","base_uri":"http://localhost:8082/consumers/my_avro_consumer/instances/my_consumer_instance"}
   $ curl -X GET -H "Accept: application/vnd.kafka.avro.v1+json" \
         http://localhost:8082/consumers/my_avro_consumer/instances/my_consumer_instance/topics/avrotest
     [{"key":null,"value":{"name":"testUser"},"partition":0,"offset":0}]
   $ curl -X DELETE \
         http://localhost:8082/consumers/my_avro_consumer/instances/my_consumer_instance
     # No content in response

Produce and Consume JSON Messages
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. sourcecode:: bash

   # Produce a message using JSON with the value '{ "foo": "bar" }' to the topic test
   $ curl -X POST -H "Content-Type: application/vnd.kafka.json.v1+json" \
         --data '{"records":[{"value":{"foo":"bar"}}]}' "http://localhost:8082/topics/jsontest"
     {"offsets":[{"partition":0,"offset":0,"error_code":null,"error":null}],"key_schema_id":null,"value_schema_id":null}

   # Create a consumer for JSON data, starting at the beginning of the topic's
   # log. Then consume some data from a topic using the base URL in the first response.
   # Finally, close the consumer with a DELETE to make it leave the group and clean up
   # its resources.
   $ curl -X POST -H "Content-Type: application/vnd.kafka.v1+json" \
         --data '{"name": "my_consumer_instance", "format": "json", "auto.offset.reset": "smallest"}' \
         http://localhost:8082/consumers/my_json_consumer
     {"instance_id":"my_consumer_instance",
     "base_uri":"http://localhost:8082/consumers/my_json_consumer/instances/my_consumer_instance"}
   $ curl -X GET -H "Accept: application/vnd.kafka.json.v1+json" \
         http://localhost:8082/consumers/my_json_consumer/instances/my_consumer_instance/topics/jsontest
     [{"key":null,"value":{"foo":"bar"},"partition":0,"offset":0}]
   $ curl -X DELETE \
         http://localhost:8082/consumers/my_json_consumer/instances/my_consumer_instance
     # No content in response

Produce and Consume Binary Messages
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. sourcecode:: bash

   # Produce a message using binary embedded data with value "Kafka" to the topic test
   $ curl -X POST -H "Content-Type: application/vnd.kafka.binary.v1+json" \
         --data '{"records":[{"value":"S2Fma2E="}]}' "http://localhost:8082/topics/binarytest"
     {"offsets":[{"partition":0,"offset":0,"error_code":null,"error":null}],"key_schema_id":null,"value_schema_id":null}

   # Create a consumer for binary data, starting at the beginning of the topic's
   # log. Then consume some data from a topic using the base URL in the first response.
   # Finally, close the consumer with a DELETE to make it leave the group and clean up
   # its resources.
   $ curl -X POST -H "Content-Type: application/vnd.kafka.v1+json" \
         --data '{"name": "my_consumer_instance", "format": "binary", "auto.offset.reset": "smallest"}' \
         http://localhost:8082/consumers/my_binary_consumer
     {"instance_id":"my_consumer_instance","base_uri":"http://localhost:8082/consumers/my_binary_consumer/instances/my_consumer_instance"}
   $ curl -X GET -H "Accept: application/vnd.kafka.binary.v1+json" \
         http://localhost:8082/consumers/my_binary_consumer/instances/my_consumer_instance/topics/binarytest
     [{"key":null,"value":"S2Fma2E=","partition":0,"offset":0}]
   $ curl -X DELETE \
         http://localhost:8082/consumers/my_binary_consumer/instances/my_consumer_instance
     # No content in response

Features
--------

Eventually, the REST Proxy should be able to expose all of the functionality
of the Java producers, consumers, and command-line tools. Here is the list of
what is currently supported:<commit_msg>fix typo in intro.<commit_after>.. _kafkarest_intro:

Kafka REST Proxy
================

The Kafka REST Proxy provides a RESTful interface to a Kafka cluster. It makes
it easy to produce and consume messages, view the state of the cluster, and
perform administrative actions without using the native Kafka protocol or
clients. Examples of use cases include reporting data to Kafka from any
frontend app built in any language, ingesting messages into a stream processing
framework that doesn't yet support Kafka, and scripting administrative actions.

Quickstart
----------

Start by running the REST Proxy and the services it depends on: ZooKeeper, Kafka, and the Schema
Registry:

.. sourcecode:: bash

   $ ./bin/zookeeper-server-start ./etc/kafka/zookeeper.properties &
   $ ./bin/kafka-server-start ./etc/kafka/server.properties &
   $ ./bin/schema-registry-start ./etc/schema-registry/schema-registry.properties &
   $ ./bin/kafka-rest-start ./etc/kafka-rest/kafka-rest.properties &

.. ifconfig:: platform_docs

   See the :ref:`Confluent Platform quickstart<quickstart>` for a more detailed explanation of how
   to get these services up and running.

Inspect Topic Metadata
~~~~~~~~~~~~~~~~~~~~~~

.. sourcecode:: bash

   # Get a list of topics
   $ curl "http://localhost:8082/topics"
     ["test","test2","test3"]

   # Get info about one topic
   $ curl "http://localhost:8082/topics/test"
     {"test":"connect-test","configs":{},"partitions":[{"partition":0,"leader":0,"replicas":[{"broker":0,"leader":true,"in_sync":true}]},{"partition":1,"leader":0,"replicas":[{"broker":1,"leader":true,"in_sync":true}]}]}

   # Get info about a topic's partitions
   $ curl "http://localhost:8082/topics/test/partitions"
     [{"partition":0,"leader":0,"replicas":[{"broker":0,"leader":true,"in_sync":true}]},{"partition":1,"leader":0,"replicas":[{"broker":1,"leader":true,"in_sync":true}]}]

Produce and Consume Avro Messages
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. sourcecode:: bash

   # Produce a message using Avro embedded data, including the schema which will
   # be registered with the schema registry and used to validate and serialize
   # before storing the data in Kafka
   $ curl -X POST -H "Content-Type: application/vnd.kafka.avro.v1+json" \
         --data '{"value_schema": "{\"type\": \"record\", \"name\": \"User\", \"fields\": [{\"name\": \"name\", \"type\": \"string\"}]}", "records": [{"value": {"name": "testUser"}}]}' \
         "http://localhost:8082/topics/avrotest"
     {"offsets":[{"partition":0,"offset":0,"error_code":null,"error":null}],"key_schema_id":null,"value_schema_id":21}

   # Create a consumer for Avro data, starting at the beginning of the topic's
   # log. Then consume some data from a topic, which is decoded, translated to
   # JSON, and included in the response. The schema used for deserialization is
   # fetched automatically from the schema registry. Finally, clean up.
   $ curl -X POST -H "Content-Type: application/vnd.kafka.v1+json" \
         --data '{"name": "my_consumer_instance", "format": "avro", "auto.offset.reset": "smallest"}' \
         http://localhost:8082/consumers/my_avro_consumer
     {"instance_id":"my_consumer_instance","base_uri":"http://localhost:8082/consumers/my_avro_consumer/instances/my_consumer_instance"}
   $ curl -X GET -H "Accept: application/vnd.kafka.avro.v1+json" \
         http://localhost:8082/consumers/my_avro_consumer/instances/my_consumer_instance/topics/avrotest
     [{"key":null,"value":{"name":"testUser"},"partition":0,"offset":0}]
   $ curl -X DELETE \
         http://localhost:8082/consumers/my_avro_consumer/instances/my_consumer_instance
     # No content in response

Produce and Consume JSON Messages
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. sourcecode:: bash

   # Produce a message using JSON with the value '{ "foo": "bar" }' to the topic jsontest
   $ curl -X POST -H "Content-Type: application/vnd.kafka.json.v1+json" \
         --data '{"records":[{"value":{"foo":"bar"}}]}' "http://localhost:8082/topics/jsontest"
     {"offsets":[{"partition":0,"offset":0,"error_code":null,"error":null}],"key_schema_id":null,"value_schema_id":null}

   # Create a consumer for JSON data, starting at the beginning of the topic's
   # log. Then consume some data from a topic using the base URL in the first response.
   # Finally, close the consumer with a DELETE to make it leave the group and clean up
   # its resources.
   $ curl -X POST -H "Content-Type: application/vnd.kafka.v1+json" \
         --data '{"name": "my_consumer_instance", "format": "json", "auto.offset.reset": "smallest"}' \
         http://localhost:8082/consumers/my_json_consumer
     {"instance_id":"my_consumer_instance",
     "base_uri":"http://localhost:8082/consumers/my_json_consumer/instances/my_consumer_instance"}
   $ curl -X GET -H "Accept: application/vnd.kafka.json.v1+json" \
         http://localhost:8082/consumers/my_json_consumer/instances/my_consumer_instance/topics/jsontest
     [{"key":null,"value":{"foo":"bar"},"partition":0,"offset":0}]
   $ curl -X DELETE \
         http://localhost:8082/consumers/my_json_consumer/instances/my_consumer_instance
     # No content in response

Produce and Consume Binary Messages
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. sourcecode:: bash

   # Produce a message using binary embedded data with value "Kafka" to the topic binarytest
   $ curl -X POST -H "Content-Type: application/vnd.kafka.binary.v1+json" \
         --data '{"records":[{"value":"S2Fma2E="}]}' "http://localhost:8082/topics/binarytest"
     {"offsets":[{"partition":0,"offset":0,"error_code":null,"error":null}],"key_schema_id":null,"value_schema_id":null}

   # Create a consumer for binary data, starting at the beginning of the topic's
   # log. Then consume some data from a topic using the base URL in the first response.
   # Finally, close the consumer with a DELETE to make it leave the group and clean up
   # its resources.
   $ curl -X POST -H "Content-Type: application/vnd.kafka.v1+json" \
         --data '{"name": "my_consumer_instance", "format": "binary", "auto.offset.reset": "smallest"}' \
         http://localhost:8082/consumers/my_binary_consumer
     {"instance_id":"my_consumer_instance","base_uri":"http://localhost:8082/consumers/my_binary_consumer/instances/my_consumer_instance"}
   $ curl -X GET -H "Accept: application/vnd.kafka.binary.v1+json" \
         http://localhost:8082/consumers/my_binary_consumer/instances/my_consumer_instance/topics/binarytest
     [{"key":null,"value":"S2Fma2E=","partition":0,"offset":0}]
   $ curl -X DELETE \
         http://localhost:8082/consumers/my_binary_consumer/instances/my_consumer_instance
     # No content in response

Features
--------

Eventually, the REST Proxy should be able to expose all of the functionality
of the Java producers, consumers, and command-line tools. Here is the list of
what is currently supported:
{'fraction_non_alphanumeric': 0.16607414931437278, 'fraction_numerical': 0.01639701081041863, 'mean_word_length': 4.307662687716596, 'pattern_counts': {'":': 184, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '176163', 'n_tokens_mistral': 4438, 'n_tokens_neox': 4159, 'n_words': 1092}