如何访问合流kafka rest代理中主题的最新偏移量以计算延迟

Question

在 confluent kafka rest 代理中，我们可以获取特定消费者组的最后一个提交偏移量，但是我们如何获取主题的最新偏移量来计算滞后。

Answer 1

您可以使用Kafka REST代理获取为特定分区提交的最新偏移量。根据汇合的文件，

GET /consumers/(string: group_name)/instances/(string: instance)/offsets

获取给定分区的上次提交的偏移量（无论提交是通过此过程还是其他进程发生的）。

请注意，必须向保存使用者实例的特定 REST 代理实例发出此请求。

参数：

> < li>

组名(字符串)-使用者组的名称

实例（字符串）-- 使用者实例请求 JSON 的 ID

对象阵列：

• partitions—查找
最后提交的偏移量的分区列表
• 分区[i]。topic（string）--主题的名称
• 分区[i]。partition（int）--分区ID

响应JSON对象数组：

• 偏移量 -- 已提交的偏移量列表
• offsets[i].topic （string） -- 为其提交偏移量的主题的名称
• offsets[i].partition （int） -- 为其提交偏移量的分区 ID
• 偏移量[i].offset （int） -- 已提交的偏移量
• 偏移量[i].元数据（字符串）-- 已提交偏移量的元数据

状态代码：

• 404 未找到 --
• 错误代码 40402 -- 找不到分区
• 错误代码 40403 -- 找不到使用者实例

请求示例:

GET /consumers/testgroup/instances/my_consumer/offsets HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Accept: application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json

{
  "partitions": [
    {
      "topic": "test",
      "partition": 0
    },
    {
      "topic": "test",
      "partition": 1
    }

  ]
}

示例响应：

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{"offsets":
 [
  {
    "topic": "test",
    "partition": 0,
    "offset": 21,
    "metadata":""
  },
  {
    "topic": "test",
    "partition": 1,
    "offset": 31,
    "metadata":""
  }
 ]
}