这是来自confluent的RESTAPI（rest代理）代码示例。我得把它打出来，所以可能有拼写错误。我希望这对你有点帮助。
（生产者使用用python编写的restapi）

import requests
import base64
import json
url = "http://restproxy:8082/topics/my_topic"
headers = {
"Content-Type" : "application/vnd.kafka.binary.v1 + json"
   }
# Create one or more messages
payload = {"records":
       [{
           "key":base64.b64encode("firstkey"),
           "value":base64.b64encode("firstvalue")
       }]}
# Send the message
r = requests.post(url, data=json.dumps(payload), headers=headers)
if r.status_code != 200:
   print "Status Code: " + str(r.status_code)
   print r.text

（使用用python编写的restapi的使用者）

import requests
import base64
import json
import sys
# Base URL for interacting with REST server
baseurl = "http://restporxy:8082/consumers/group1"
# Create the Consumer instance
print "Creating consumer instance"
payload {
    "format": "binary"
    }
headers = {
"Content-Type" : "application/vnd.kafka.v1+json"
    }
r = requests.post(baseurl, data=json.dumps(payload), headers=headers)
if r.status_code !=200:
    print "Status Code: " + str(r.status_code)
    print r.text
    sys.exit("Error thrown while creating consumer")
# Base URI is used to identify the consumer instance
base_uri = r.json()["base_uri"]
# Get the messages from the consumer
headers = {
    "Accept" : "application/vnd.kafka.binary.v1 + json"
    }
# Request messages for the instance on the Topic
r = requests.get(base_uri + "/topics/my_toopic", headers = headers, timeout =20)
if r.status_code != 200: 
    print "Status Code: " + str(r.status_code)
    print r.text
    sys.exit("Error thrown while getting message")
# Output all messages
for message in r.json():
    if message["key"] is not None:
        print "Message Key:" + base64.b64decode(message["key"])
    print "Message Value:" + base64.b64decode(message["value"])
# When we're done, delete the consumer
headers = {
    "Accept" : "application/vnd.kafka.v1+json"
    }
r = requests.delete(base_uri, headers=headers)
if r.status_code != 204: 
    print "Status Code: " + str(r.status_code)
    print r.text

我猜您的问题更多的是关于如何编写“服务器”restapi接口，而不是客户端（客户端最后只是发出http请求）。例如，可以使用strimzihttp桥(https://github.com/strimzi/strimzi-kafka-bridge)在kubernetes中，哪一个是独立工作的，哪一个是您愿意在那里部署集群的（然后您可以使用strimzi项目，https://strimzi.io/).