kafka-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Dana Powers <dana.pow...@gmail.com>
Subject Re: Resetting offsets
Date Thu, 04 May 2017 03:52:47 GMT
Requires stopping your existing consumers, but otherwise should work:

from kafka import KafkaConsumer, TopicPartition, OffsetAndMetadata

def reset_offsets(group_id, topic, bootstrap_servers):
  consumer = KafkaConsumer(bootstrap_servers=bootstrap_servers,
group_id=group_id)
  consumer.assign([TopicPartition(topic, i) for i in
consumer.partitions_for_topic(topic)])
  consumer.commit({tp: OffsetAndMetadata(0, b'') for tp in
consumer.assignment()})
  consumer.close()

On May 3, 2017 9:20 AM, "Ben Stopford" <ben@confluent.io> wrote:

> Hu is correct, there isn't anything currently, but there is an active
> proposal:
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> 122%3A+Add+Reset+Consumer+Group+Offsets+tooling
>
> On Wed, May 3, 2017 at 1:23 PM Hu Xi <huxi_2b@hotmail.com> wrote:
>
> > Seems there is no command line out of box, but if you could write a
> simple
> > Java client application that firstly calls 'seek' or 'seekToBeginning' to
> > reset offsets to what you expect and then invoke commitSync to commit the
> > offsets.
> >
> >
> > ________________________________
> > 发件人: Paul van der Linden <paul@sportr.co.uk>
> > 发送时间: 2017年5月3日 18:28
> > 收件人: users@kafka.apache.org
> > 主题: Resetting offsets
> >
> > I'm trying to reset the offsets for all partitions for all topics for a
> > consumer group, but I can't seem to find a working way.
> >
> > The command line tool provides a tool to remove a consumer group (which
> > would be fine in this occasion), but this is not working with the "new"
> > style consumer groups. I tried to set consumer offsets with a client,
> which
> > also didn't work (we are using confluent-kafka-python with librdkafka).
> >
> > Is there any way to reset the offsets (preferable with python or a
> command
> > line tool)?
> >
> > Thanks
> >
>

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message