Support Kafka's ConsumerConfig.MAX_POLL_RECORDS_CONFIG to config max number of message will return in a single poll

[MarvinCai]

Motivation
Support Kafka's ConsumerConfig.MAX_POLL_RECORDS_CONFIG to config max number of message will return in a single poll. #1090
Also update doc to reflect that we already supporting earlist and latest strategy for ConsumerConfig.AUTO_OFFSET_RESET_CONFIG.

Kafka config parser has check to make sure 'max.poll.records' value is greater than 1 and will throw exception if config < 1, so I didn't add extra check.

[merlimat]

we should still use the constant for 1000, maybe renaming to DEFAULT_MAX_RECORDS_IN_SINGLE_POLL

[merlimat]

That would be the same behavior as Kafka, no?

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.

[MarvinCai]

Hi @merlimat, so for the auto.offset.reset setting, Pulsar only use it when offset info somehow not found in the offset map, so it’s not used exactly the same as it used in Kafka right?

[merlimat]

@MarvinCai Isn't the same behavior from Kafka ?

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):

[MarvinCai]

@merlimat Yes, for earliest and latest we have the same behavior, but we're not handling none or anything else case, I think we should add it.
And in the PulsarKafkaConsumer.java use ConsumerConfig.GROUP_ID_CONFIG, ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, ConsumerConfig.AUTO_OFFSET_RESET_CONFIG. But in the doc we only say we support them, which sound like optional config. So I think we should either mark them as required in the doc or provided some default value to them and specify in the doc. Which one you think would be better.

[merlimat]

but we're not handling none or anything else case, I think we should add it.

Oh I see. Very good point. The question is that I don't think we have a direct way right now to do that (other that using REST API to check the topic stats... which I don't think would be a good option).

[merlimat]

run java8 tests
run integration tests

[merlimat]

👍

[MarvinCai]

run integration tests