Confusing behaviour in Consumers Rest Proxy

[nikosgpet]

I am testing out the Consumers Rest API and I have a hard time getting the result I was looking for. It might be my fault or a bug, not sure.

Reproduction

I am running redpanda in a single cluster installation in the same docker-compose with pandaproxy.

I am testing the three following endpoints:

• GET /consumers/{consumer_group}/instances/{instance}/offsets
• POST /consumers/{consumer_group}/instances/{instance}/offsets
• GET /consumers/{consumer_group}/instances/{instance}/records

I add Content-Type: application/vnd.kafka.v2+json to all three.

In the first and third request I send the following body as well:

{
  "partitions": [
    {
      "topic": "nameoftopic",
      "partition": 0
    }
  ]
}

Issues

Firstly, all consumer endpoinds ask for instance.id, and I had a really hard time finding what my instance.id is until I realised that I was not setting a client.id for my consumer. Even after I set an id, when I execute GET /consumers/benthos_consumer_group/instances/benthos_kafka_input/offsets I keep getting:

{
    "error_code": 40403,
    "message": "{benthos_consumer_group}, {benthos_kafka_input}, { error_code: unknown_member_id [25] }"
}

With GET /consumers/benthos_consumer_group/instances/benthos_kafka_input/records I get:

{
    "error_code": 406,
    "message": "HTTP 406 Not Acceptable"
}

And with POST /consumers/benthos_consumer_group/instances/benthos_kafka_input/offsets I get:

{
    "message": "Not found",
    "code": 404
}

Use case

I am using kafka to build data pipelines, and sometimes I want the consumer to re-consume a topic from the beginning (for ex. if I made a change in the pipeline, or there was an error in the consumer). I was searching for a way to manually reset/change the offset of a consumer group for a particular topic so that it re-consumes part of / all past events.

This scenario is also typical in event-driven programming, where re-consuming past events is a recurring use case.

To circumvent these issues, I created a temporary consumer for my pipeline until it re-consumed everything. It works, but it's not ideal.

Feedback:

• Since instance.id are necessary, I think it would be great to be able to do a simple GET to list all available consumer groups & instances per topic. Otherwise, it's like looking for a needle in the haystack.
• It would be really cool if we could reset the offset of a consumer for a particular topic using rpk

[emaxerrno]

thanks for the report @nikosgpet !

[BenPope]

Thanks for the feedback @nikosgpet

Are you trying to manage offsets for arbitrary consumers?

Pandaproxy isn't a generalised admin tool for offset management, it only allows management of offsets for consumers that are created with it.

When a consumer is created, you can specify a name, or one will be generated and returned, along with the endpoint to connect tp: https://vectorized.io/blog/pandaproxy/#Create-a-consumer

Those are both required, because the rest proxy is stateful, storing various offsets, subscriptions, and partition assignments in the particular instance of Pandaproxy that the consumer instance was created on.

For many GET requests, you must specify an Accept: application/vnd.kafka.v2+json header: https://vectorized.io/blog/pandaproxy/#Consume-messages

If I've misunderstood, and you're having problems with consumers created on the proxy, then I'll take another look tomorrow, to make sure everything is working as expected.

[nikosgpet]

@BenPope thanks for the detailed answer! I hadn't realized that the rest proxy allows only the manipulation of those consumers created by the proxy itself. This makes more sense now.

What way would you currently recommend to manage offsets in a generalized way, for arbitrary consumers?

I find that offset management would make a neat feature for the rpk.

[BenPope]

@nikosgpet I don't have any concrete suggestions. I'm not sure there's a general way to do it, I think you'd have to shutdown the consumers, then commit offsets, and start the consumers up again.

Otherwise the request would fail the member id check, or the group generation check.

I'm not entirely sure of joining an "admin consumer" to the same group, and committing offsets would force the other consumers to re-read the offsets, but it might force a rejoin if the generation is changed, and that might force them to re-read offsets. I'm no expert here.

[nikosgpet]

For info, what I wanted to do is doable with kafka's official kafka-consumer-groups.sh. I think this functionality will be a great addition one day for the rpk.

[emaxerrno]

@twmb and @0x5d are already working on group metadata manip like the Kafka consumer group script in rpk.

I think @dotnwat files a tix for it recently