[Bug][KafkaSource]Fix commit kafka offset bug.

[lightzhao]

After the checkpoint is completed, the offset will be submitted to kafka. The offset of the submitted partition should be KafkaSourceSplit:: getStartOffset. Otherwise, some offsets are not submitted successfully, resulting in repeated data consumption.

Purpose of this pull request

Check list

• Code changed are covered with tests, or it does not need tests for reason:
• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/incubator-seatunnel/tree/dev/docs
• If you are contributing the connector code, please check that the following files are updated:
  1. Update change log that in connector document. For more details you can refer to connector-v2
  2. Update plugin-mapping.properties and add new connector information in it
  3. Update the pom file of seatunnel-dist
• Update the release-note.

[lightzhao]

@Hisoka-X @TyrantLucifer PTAL.

[lightzhao]

kafka2hdfs test case:
HdfsSink:

KafkaSource:



After the checkpoint is completed, the offset is not submitted.

[Hisoka-X]

Hi, Please check KafkaSourceReader::pollNext, the endOffset value will be increase by consumer meesage from kafka. Means the endOffset always the last message offset consume from kafka.

By the way, the offset not commit into kafka may be other reason. Because if we commit endOffset or startOffset the kafka both should commit.

[Hisoka-X]

I will close this PR, if you have any other suggestion or problem, please reopen it. Thanks!

[lightzhao]

By the way, the offset not commit into kafka may be other reason. Because if we commit endOffset or startOffset the kafka both should commit.

@Hisoka-X Offset starts from 0. When submitting, it should be +1. Otherwise, the lag is always missing 1. I have verified it in the test environment. You can also try it.

[lightzhao]

@Hisoka-X hello, I don't seem to have permission to reopen this PR. Can you reopen it? My test is really a bug. You can test it here。

[Hisoka-X]

By the way, the offset not commit into kafka may be other reason. Because if we commit endOffset or startOffset the kafka both should commit.

@Hisoka-X Offset starts from 0. When submitting, it should be +1. Otherwise, the lag is always missing 1. I have verified it in the test environment. You can also try it.

I got your point.

[Hisoka-X]

It is indeed a bug, thank you for your unremitting efforts