Closed xelax closed 10 years ago
hi @xelax, so the default for kafka consumers for auto.offset.reset
is largest
, which is how the akka-kafka consumer behaves.
to override this behavior you can add this to your application.conf
used to configure the actor system you use with your AkkaConsumer
kafka.consumer.auto.offset.reset = "smallest"
Thanks for the quick answer! I noticed that zookeeper it will remember the offset for my consumer group, so the setting operates only when there is no offset stored in zookeper for my consumer group. How do I "re-start", meaning, tell kafka to ignore the stored offset and start anew from the newest message? It would be useful, particularly for testing purpose.
So the way kafka's console consumer works is that if you do --from-beginning
it will just delete the consumer offsets from zookeeper before starting.
So for testing you can do similar in a before
method, like
val props:AkkaConsumerProps = //your consumer props
ZkUtils.maybeDeletePath(props.zkConnect , "/consumers/" + props.group)
also ZkUtils
is kafka.utils.ZkUtils
which should already be on your classpath if akka-kafka
is
How do I tell the consumer whether to start from the oldest available message or from the newest available offset?