confluentinc/kafka-rest confluentinc/kafka-rest

  • Stars
    star
    2,137
  • Rank 21,593 (Top 0.5 %)
  • Language
    Java
  • License
    Other
  • Created about 10 years ago
  • Updated about 1 year ago
Twitter logo Share LinkedIn logo Share Facebook logo Share
confluentinc/kafka-rest
github

confluentinc

Reviews

There are no reviews yet. Be the first to send feedback to the community and the maintainers!

Repository Details

Confluent REST Proxy for Kafka

Kafka REST Proxy

The Kafka REST Proxy provides a RESTful interface to a Kafka cluster. It makes it easy to produce and consume data, 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 front-end app built in any language, ingesting data into a stream processing framework that doesn't yet support Kafka, and scripting administrative actions.

Installation

You can download prebuilt versions of the Kafka REST Proxy as part of the Confluent Platform.

You can read our full installation instructions and the complete documentation.

To install from source, follow the instructions in the Development section below.

Deployment

The Kafka REST Proxy includes a built-in Jetty server and can be deployed after being configured to connect to an existing Kafka cluster.

Running mvn clean package runs all 3 of its assembly targets.

  • The development target assembles all necessary dependencies in a kafka-rest/target subfolder without packaging them in a distributable format. The wrapper scripts bin/kafka-rest-start and bin/kafka-rest-stop can then be used to start and stop the service.
  • The package target is meant to be used in shared dependency environments and omits some dependencies expected to be provided externally. It assembles the other dependencies in a kafka-rest/target subfolder as well as in distributable archives. The wrapper scripts bin/kafka-rest-start and bin/kafka-rest-stop can then be used to start and stop the service.
  • The standalone target packages all necessary dependencies as a distributable JAR that can be run as standard (java -jar $base-dir/kafka-rest/target/kafka-rest-X.Y.Z-standalone.jar).

Quickstart (v3 API)

The following assumes you have Kafka and an instance of the REST Proxy running using the default settings and some topics already created.

The v3 API is the latest version of the API. The cluster ID is a path parameter to enable a REST Proxy to work with multiple Kafka clusters. API responses often contain links to related resources, such as the list of a topic's partitions. The content type is always application/json.

Get the local cluster information

$ curl http://localhost:8082/v3/clusters

Response:
  {"kind":"KafkaClusterList",
   "metadata":{"self":"http://localhost:8082/v3/clusters","next":null},
   "data":[
    {"kind":"KafkaCluster",
     "metadata":{"self":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q",
     "resource_name":"crn:///kafka=xFhUvurESIeeCI87SXWR-Q"},
     "cluster_id":"xFhUvurESIeeCI87SXWR-Q",
     "controller":{"related":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/brokers/0"},
     "acls":{"related":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/acls"},
     "brokers":{"related":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/brokers"},
     "broker_configs":{"related":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/broker-configs"},
     "consumer_groups":{"related":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/consumer-groups"},
     "topics":{"related":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics"},
     "partition_reassignments":{"related":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics/-/partitions/-/reassignment"}
    }
   ]
  }

The cluster ID in the output is xFhUvurESIeeCI87SXWR-Q.

Get a list of topics

$ curl http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics

Response:
  {"kind":"KafkaTopicList",
   "metadata":{"self":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics","next":null},
   "data":[
    {"kind":"KafkaTopic",
     "metadata":{"self":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics/jsontest",
     "resource_name":"crn:///kafka=xFhUvurESIeeCI87SXWR-Q/topic=jsontest"},
     "cluster_id":"xFhUvurESIeeCI87SXWR-Q",
     "topic_name":"jsontest",
     "is_internal":false,
     "replication_factor":1,
     "partitions_count":1,
     "partitions":{"related":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics/jsontest/partitions"},
     "configs":{"related":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics/jsontest/configs"},
     "partition_reassignments":{"related":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics/jsontest/partitions/-/reassignment"}
    }
   ]
  }

Create a topic

$ curl -X POST -H "Content-Type:application/json" -d '{"topic_name":"jsontest"}' \
       http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics

Response:
  {"kind":"KafkaTopic",
   "metadata":{"self":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics/jsontest",
   "resource_name":"crn:///kafka=xFhUvurESIeeCI87SXWR-Q/topic=jsontest"},
   "cluster_id":"xFhUvurESIeeCI87SXWR-Q",
   "topic_name":"jsontest",
   "is_internal":false,
   "replication_factor":1,
   "partitions_count":1,
   "partitions":{"related":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics/jsontest/partitions"},
   "configs":{"related":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics/jsontest/configs"},
   "partition_reassignments":{"related":"http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics/jsontest/partitions/-/reassignment"}
  }

Produce records with JSON data

$ curl -X POST -H "Content-Type: application/json" \
       -d '{"value":{"type":"JSON","data":{"name":"testUser"}}}' \
       http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics/jsontest/records

Response:
  {"error_code":200,
   "cluster_id":"xFhUvurESIeeCI87SXWR-Q",
   "topic_name":"jsontest",
   "partition_id":0,
   "offset":0,
   "timestamp":"2023-03-09T14:07:23.592Z",
   "value":{"type":"JSON","size":19}
  }

In the response, the error_code of 200 is an HTTP status code (OK) which indicates the operation was successful. Because you can use this API to stream multiple records into a topic as part of the same request, each record produced has its own error code. To send multiple records, simply concatentate the records like this:

$ curl -X POST -H "Content-Type: application/json" \
       -d '{"value":{"type":"JSON","data":"ONE"}} {"value":{"type":"JSON","data":"TWO"}}' \
       http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics/jsontest/records

Response:
  {"error_code":200,
   "cluster_id":"xFhUvurESIeeCI87SXWR-Q",
   "topic_name":"jsontest",
   "partition_id":0,
   "offset":1,
   "timestamp":"2023-03-09T14:07:23.592Z",
   "value":{"type":"JSON","size":5}
  }
  {"error_code":200,
   "cluster_id":"xFhUvurESIeeCI87SXWR-Q",
   "topic_name":"jsontest",
   "partition_id":0,
   "offset":2,
   "timestamp":"2023-03-09T14:07:23.592Z",
   "value":{"type":"JSON","size":5}
  }

Produce records with string data

$ curl -X POST -H "Content-Type: application/json" \
       -d '{"value":{"type":"STRING","data":"REST"}}' \
       http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics/jsontest/records

Response:
  {"error_code":200,
   "cluster_id":"xFhUvurESIeeCI87SXWR-Q",
   "topic_name":"jsontest",
   "partition_id":0,
   "offset":2,
   "timestamp":"2023-03-09T14:07:23.592Z",
   "value":{"type":"STRING","size":4}
  }

The data is treated as a string in UTF-8 encoding and follows JSON rules for escaping special characters.

Produce records in a batch

As an alternative to streaming mode, you can produce multiple records in a batch. This is not streaming, but it is easier to use with HTTP libraries that expect a straightforward request-response behavior.

Each entry in the batch has a unique identifier (a string of up to 80 characters) which can be used to correlate the responses. The identifiers of the entries in a batch must be unique.

$ curl -X POST -H "Content-Type: application/json" \
       -d '{"entries":[{"id":"first","value":{"type":"JSON","data":"ONE"}}, {"id":"second","value":{"type":"JSON","data":"TWO"}}]}' \
       http://localhost:8082/v3/clusters/xFhUvurESIeeCI87SXWR-Q/topics/jsontest/records:batch

Response:
  {"successes":[
    {"id":"first",
     "cluster_id":"xFhUvurESIeeCI87SXWR-Q",
     "topic_name":"jsontest",
     "partition_id":0,
     "offset":3,
     "timestamp":"2023-03-09T14:07:23.592Z",
     "value":{"type":"JSON","size":5}
    },
    {"id":"second",
     "cluster_id":"xFhUvurESIeeCI87SXWR-Q",
     "topic_name":"jsontest",
     "partition_id":0,
     "offset":4,
     "timestamp":"2023-03-09T14:07:23.592Z",
     "value":{"type":"JSON","size":5}
    }
   ],
   "failures":[]
  }

Successes and failures are returned in the response in separate arrays like this:

{
  "successes": [
    {
      "id": "1",
      "cluster_id": "xFhUvurESIeeCI87SXWR-Q",
      "topic_name": "jsontest",
      "partition_id": 0,
      "offset": 5,
      "timestamp": "2023-03-09T14:07:23.592Z",
      "value": {
        "type": "STRING",
        "size": 7
      }
    }
  ],
  "failures": [
    {
      "id": "2",
      "error_code": 400,
      "message": "Bad Request: data=\"Message$\" is not a valid base64 string."
    }
  ]
}

Quickstart (v2 API)

The earlier v2 API is a bit more concise.

Get a list of topics

$ curl http://localhost:8082/topics
  
Response:
  ["__consumer_offsets","jsontest"]

Get info about one topic

$ curl http://localhost:8082/topics/jsontest

Response:
  {"name":"jsontest",
   "configs":{},
   "partitions":[
    {"partition":0,
     "leader":0,
     "replicas":[
      {"broker":0,
       "leader":true,
       "in_sync":true
      }
     ]
    }
   ]
  }

Produce records with JSON data

$ curl -X POST -H "Content-Type: application/vnd.kafka.json.v2+json" \
       -d '{"records":[{"value":{"name": "testUser"}}]}' \
       http://localhost:8082/topics/jsontest

Response:
  {"offsets":[
    {"partition":0,
     "offset":0,
     "error_code":null,
     "error":null
    }
   ],
   "key_schema_id":null,
   "value_schema_id":null
  }

Consume JSON data

First, create a consumer for JSON data, starting at the beginning of the topic. The consumer group is called my_json_consumer and the instance is my_consumer_instance.

$ curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" -H "Accept: application/vnd.kafka.v2+json" \
       -d '{"name": "my_consumer_instance", "format": "json", "auto.offset.reset": "earliest"}' \
       http://localhost:8082/consumers/my_json_consumer

Response:
  {"instance_id":"my_consumer_instance",
   "base_uri":"http://localhost:8082/consumers/my_json_consumer/instances/my_consumer_instance"
  }

Subscribe the consumer to a topic.

$ curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" \
       -d '{"topics":["jsontest"]}' \
      http://localhost:8082/consumers/my_json_consumer/instances/my_consumer_instance/subscription

Response:
  # No content in response

Then consume some data from a topic using the base URL in the first response.

$ curl -X GET -H "Accept: application/vnd.kafka.json.v2+json" \
       http://localhost:8082/consumers/my_json_consumer/instances/my_consumer_instance/records

Response:
  [
   {"key":null,
    "value":{"name":"testUser"},
    "partition":0,
    "offset":0,
    "topic":"jsontest"
   }
  ]

Finally, close the consumer with a DELETE to make it leave the group and clean up its resources.

$ curl -X DELETE -H "Accept: application/vnd.kafka.v2+json" \
       http://localhost:8082/consumers/my_json_consumer/instances/my_consumer_instance

Response:
  # No content in response

Development

To build a development version, you may need development versions of common, rest-utils, and schema-registry. After installing these, you can build the Kafka REST Proxy with Maven. All the standard lifecycle phases work.

You can avoid building development versions of dependencies by building on the latest (or earlier) release tag, or <release>-post branch, which will reference dependencies available pre-built from the public repository. For example, branch 7.3.0-post can be used as a base for patches for this version.

Contribute

License

This project is licensed under the Confluent Community License.

More Repositories

1

ksql

The database purpose-built for stream processing applications.
Java
5,533
star
2

confluent-kafka-go

Confluent's Apache Kafka Golang client
Go
4,402
star
3

confluent-kafka-python

Confluent's Kafka Python Client
Python
3,388
star
4

confluent-kafka-dotnet

Confluent's Apache Kafka .NET client
C#
2,560
star
5

kafka-streams-examples

Demo applications and code examples for Apache Kafka's Streams API.
Java
2,169
star
6

schema-registry

Confluent Schema Registry for Kafka
Java
2,022
star
7

examples

Apache Kafka and Confluent Platform examples and demos
Shell
1,903
star
8

bottledwater-pg

Change data capture from PostgreSQL into Kafka
C
1,521
star
9

demo-scene

👾Scripts and samples to support Confluent Demos and Talks. ⚠️Might be rough around the edges ;-) 👉For automated tutorials and QA'd code, see https://github.com/confluentinc/examples/
Shell
1,356
star
10

cp-docker-images

[DEPRECATED] Docker images for Confluent Platform.
Python
1,142
star
11

kafka-connect-jdbc

Kafka Connect connector for JDBC-compatible databases
Java
953
star
12

cp-all-in-one

docker-compose.yml files for cp-all-in-one , cp-all-in-one-community, cp-all-in-one-cloud, Apache Kafka Confluent Platform
Python
889
star
13

cp-helm-charts

The Confluent Platform Helm charts enable you to deploy Confluent Platform services on Kubernetes for development, test, and proof of concept environments.
Mustache
764
star
14

kafka-connect-elasticsearch

Kafka Connect Elasticsearch connector
Java
715
star
15

parallel-consumer

Parallel Apache Kafka client wrapper with per message ACK, client side queueing, a simpler consumer/producer API with key concurrency and extendable non-blocking IO processing.
Java
654
star
16

kafka-connect-hdfs

Kafka Connect HDFS connector
Java
465
star
17

kafka-tutorials

Tutorials and Recipes for Apache Kafka
Java
302
star
18

kafka-images

Confluent Docker images for Apache Kafka
Python
295
star
19

ducktape

System integration and performance tests
Python
294
star
20

librdkafka

The Apache Kafka C/C++ library
C
180
star
21

kafka-rest-node

Node.js client for the Kafka REST proxy
JavaScript
146
star
22

confluent-platform-security-tools

Security tools for the Confluent Platform.
Shell
146
star
23

kafka-connect-datagen

Connector that generates data for demos
Java
143
star
24

docker-images

DEPRECATED - Dockerfiles for Confluent Stream Data Platform
Shell
116
star
25

rest-utils

Utilities and a small framework for building REST services with Jersey, Jackson, and Jetty.
Java
111
star
26

cli

CLI for Confluent Cloud and Confluent Platform
Go
103
star
27

openmessaging-benchmark

Java
89
star
28

camus

Mirror of Linkedin's Camus
Java
88
star
29

common

Common utilities library containing metrics, config and utils
Java
85
star
30

kafka-workshop

JavaScript
75
star
31

confluent-kafka-javascript

Confluent's Apache Kafka JavaScript client
JavaScript
71
star
32

training-developer-src

Source Code accompanying the Confluent Kafka for Developers course
Java
70
star
33

ccloud-tools

Running Tools from Confluent Platform along with your Confluent Cloud™ Cluster
HCL
67
star
34

bincover

Easily measure code coverage of Golang binaries
Go
62
star
35

libserdes

Avro Serialization/Deserialization C/C++ library with Confluent schema-registry support
C
62
star
36

kafka-connect-blog

Demo for Kafka Connect with JDBC and HDFS Connectors
Shell
59
star
37

confluent-cli

Confluent Platform CLI
Shell
58
star
38

ksqldb-graphql

Node.js GraphQL integration for ksqlDB
TypeScript
56
star
39

confluent-sigma

JavaScript
52
star
40

terraform-provider-confluentcloud

Confluent Cloud Terraform Provider is deprecated in favor of Confluent Terraform Provider
Go
52
star
41

jmx-monitoring-stacks

📊 Monitoring examples for Confluent Cloud and Confluent Platform
C#
44
star
42

confluent-kubernetes-examples

Example scenario workflows for Confluent for Kubernetes
Shell
43
star
43

qcon-microservices

Example online orders app composed of event-driven microservices. Built for QCon workshop.
Java
38
star
44

securing-kafka-blog

Secure Kafka cluster (in a VM) for development and testing
Puppet
38
star
45

cp-demo

Confluent Platform Demo including Apache Kafka, ksqlDB, Control Center, Schema Registry, Security, Schema Linking, and Cluster Linking
Shell
36
star
46

training-administration-src

Contains docker-compose file needed for Apache Kafka Administration by Confluent training
HTML
36
star
47

mox

A hybrid mock and proxy server - easily programmable and runs on express
JavaScript
35
star
48

terraform-state-s3

Terraform module to create the S3/DynamoDB backend to store the Terraform state+lock
HCL
34
star
49

common-docker

Confluent Commons with support for building and testing Docker images.
Java
34
star
50

ksql-recipes-try-it-at-home

Files needed to try out KSQL Recipes for yourself
Shell
34
star
51

training-ksql-and-streams-src

Sample solutions for the exercises of the course KSQL & Kafka Streams
Java
30
star
52

schema-registry-images

Docker Images for Schema Registry
Python
29
star
53

terraform-provider-confluent

Terraform Provider for Confluent
Go
29
star
54

confluent-docker-utils

Common Python utils for testing Confluent's Docker images
Python
28
star
55

flink-cookbook

Java
28
star
56

cp-ansible

Ansible playbooks for the Confluent Platform
Jinja
28
star
57

ksql-images

KSQL platform docker images
Shell
27
star
58

coding-in-motion

Source code for the "Coding in Motion" series.
Nix
25
star
59

proto-go-setter

Go
23
star
60

online-inferencing-blog-application

Source code and application accompanying the online inferencing blog
Java
21
star
61

stream-me-up-scotty

A wide range of Digital Assets from Confluent's Solution Engineering team for Confluent Cloud
21
star
62

training-fundamentals-src

Source code accompanying the course "Apache Kafka Technical Essentials"
Shell
19
star
63

infoq-kafka-ksql

Code samples to go with InfoQ article
Shell
17
star
64

kafka-rest-images

Docker Images for Kafka REST
Python
17
star
65

kafka-mqtt-images

Confluent Docker images for Kafka MQTT
Shell
16
star
66

learn-kafka-courses

Learn the basics of Apache Kafka® from leaders in the Kafka community with these video courses covering the Kafka ecosystem and hands-on exercises.
Shell
16
star
67

commercial-workshops

Confluent Commercial SE Team's Demo and Workshop Repository
Python
14
star
68

training-cao-src

Source code accompanying the course "Monitoring, Troubleshooting and Tuning"
Java
13
star
69

ccloud-connectivity

Setup and testing connectivity to Confluent Cloud
Shell
13
star
70

event-streaming-patterns

A collection of Event Streaming Patterns, including problem statements, solutions, and implementation examples.
HTML
13
star
71

vscode

Confluent for Visual Studio Code
TypeScript
12
star
72

ksqldb-recipes

Makefile
12
star
73

ksql-workshop

KSQL Workshop
11
star
74

demo-stream-designer

Current 2022 Confluent Keynote Demo covering Stream Designer, Stream Catalog, and Stream Sharing.
Python
11
star
75

control-center-images

Docker images for enterprise control center images
Python
11
star
76

kafka-connect-http-demo

A demo target for running the Confluent HTTP sink connector
Java
11
star
77

castle

Castle is a test harness for Apache Kafka, Trogdor, and related projects.
Java
11
star
78

demo-change-data-capture

This demo shows how to capture data changes from relational databases (Oracle and PostgreSQL) and stream them to Confluent Cloud, use ksqlDB for real-time stream processing, send enriched data to cloud data warehouses (Snowflake and Amazon Redshift).
HCL
11
star
79

kafkacat-images

Docker Images for Kafkacat
10
star
80

confluent-kafka-go-dev

[EXPERIMENTAL] Development / WIP / exploratory / test fork of confluent-kafka-go
Go
10
star
81

confluent-hybrid-cloud-workshop

Confluent Hybrid Cloud Workshop
HCL
10
star
82

learn-practical-event-modeling

Kotlin
9
star
83

ksql-elasticsearch-demo

TSQL
8
star
84

strata-tutorials

Content for Spring 2016 Strata tutorials
Java
7
star
85

demo-database-modernization

This demo shows how to stream data to cloud databases with Confluent. It includes fully-managed connectors (Oracle CDC, RabbitMQ, MongoDB Atlas), ksqlDB/Flink SQL as stream processing engine.
HCL
7
star
86

flink-table-api-java-examples

Java Examples for running Apache Flink® Table API on Confluent Cloud
Java
6
star
87

confluent-oauth-extensions

Java
6
star
88

kafka-replicator-images

Docker images for Kafka Connect
Shell
6
star
89

etl

Code for ETL data pipelines
Python
6
star
90

operator-earlyaccess

Confluent Operator Early Access docs
6
star
91

schema-registry-workshop

JavaScript
6
star
92

learn-building-flink-applications-in-java-exercises

Java
6
star
93

demo-application-modernization

Application modernization example including Confluent Cloud, ksqlDB, Postgres, and Elasticsearch.
JavaScript
6
star
94

csid-secrets-providers

Enables use of external third-party systems for storing/retrieving key/value pairs with Confluent clusters.
Java
6
star
95

support-metrics-common

Common utilities for metrics collection of proactive support
Java
6
star
96

flink-table-api-python-examples

Python Examples for running Apache Flink® Table API on Confluent Cloud
Python
5
star
97

confluent-kafka-go-example

Example application using the confluent-kafka-go client
Go
5
star
98

learn-kafka-kraft

KRaft mode playground
Shell
5
star
99

ccloud-sdk-go-v2

SDK for interacting with Confluent Cloud
Makefile
5
star
100

streaming-ops

Simulated production environment running Kubernetes targeting Apache Kafka and Confluent components on Confluent Cloud. Managed by declarative infrastructure and GitOps.
Shell
5
star