support auto.offset.reset on the Pulsar adaptor for Apache Kafka

[macthestack]

auto.offset.reset is currently ignored

[ivankelly]

@macthestack could you link some docs on what this option is supposed to do?

[skyrocknroll]

@ivankelly https://kafka.apache.org/documentation.html#newconsumerconfigs . What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server (e.g. because that data has been deleted):
earliest: automatically reset the offset to the earliest offset
latest: automatically reset the offset to the latest offset
none: throw exception to the consumer if no previous offset is found for the consumer's group
anything else: throw exception to the consumer.

[ConcurrencyPractitioner]

@skyrocknroll So are you suggesting that pulsar should allow the user to determine the reset policy for the Kafka consumer?

[merlimat]

The Kafka wrapper was actually created before ConsumerBuilder.html.subscriptionInitialPosition() was available: http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerBuilder.subscriptionInitialPosition-org.apache.pulsar.client.api.SubscriptionInitialPosition-

It should be easy to support auto.offset.reset with it now.

[ConcurrencyPractitioner]

Just a couple of notes in regards to implementation. Looking through Kafka's code for auto.offset.reset, it might require some tweaking with Pulsar's internals as well since it requires knowledge of when an exception occurred for a missing offset and which offset to set it to.

[MarvinCai]

I think there's already support for auto.offset.reset , at least earliest and latest are supported, code reference.