[HUDI-1944] Support Hudi to read from committed offset

[veenaypatil]

What is the purpose of the pull request

This is a followup to - #3092, this change adds support to read from committed offset

Brief change log

• Update KafkaOffsetGen to read from consumer group committed offset
• Added NONE to KafkaResetOffsetStrategies

Verify this pull request

This change added tests and can be verified as follows:

• Added unit test testGetNextOffsetRangesFromGroup in TestKafkaOffsetGen

Committer checklist

• Has a corresponding JIRA in PR title & commit

• Commit message is descriptive of the change

• CI is green

• Necessary doc changes done or have another open PR

• For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.

[hudi-bot]

CI report:

• c5471c3 Azure: FAILURE

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run travis re-run the last Travis build
• @hudi-bot run azure re-run the last Azure build

[codecov-commenter]

Codecov Report

Merging #3175 (c5471c3) into master (039aeb6) will decrease coverage by 1.88%.
The diff coverage is 93.75%.

@@             Coverage Diff              @@
##             master    #3175      +/-   ##
============================================
- Coverage     46.16%   44.27%   -1.89%     
+ Complexity     5370     4603     -767     
============================================
  Files           921      825      -96     
  Lines         39953    36593    -3360     
  Branches       4288     3945     -343     
============================================
- Hits          18444    16202    -2242     
+ Misses        19630    18645     -985     
+ Partials       1879     1746     -133     

Flag	Coverage Δ
hudicli	39.95% <ø> (ø)
hudiclient	16.44% <ø> (-14.02%)	⬇️
hudicommon	47.58% <ø> (ø)
hudiflink	60.04% <ø> (+0.18%)	⬆️
hudihadoopmr	51.29% <ø> (ø)
hudisparkdatasource	67.06% <ø> (ø)
hudisync	54.05% <ø> (ø)
huditimelineservice	64.36% <ø> (ø)
hudiutilities	58.62% <93.75%> (+0.18%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...hudi/utilities/sources/helpers/KafkaOffsetGen.java	88.32% <93.75%> (+1.22%)	⬆️
...java/org/apache/hudi/sink/StreamWriteFunction.java	81.48% <0.00%> (-2.87%)	⬇️
...e/hudi/client/heartbeat/HoodieHeartbeatClient.java	68.22% <0.00%> (-0.94%)	⬇️
...ain/java/org/apache/hudi/util/FlinkClientUtil.java	10.71% <0.00%> (-0.40%)	⬇️
...ache/hudi/sink/StreamWriteOperatorCoordinator.java	71.60% <0.00%> (-0.34%)	⬇️
...va/org/apache/hudi/configuration/FlinkOptions.java	96.37% <0.00%> (ø)
.../org/apache/hudi/streamer/FlinkStreamerConfig.java	0.00% <0.00%> (ø)
...org/apache/hudi/client/HoodieFlinkWriteClient.java	0.00% <0.00%> (ø)
...udi/table/action/commit/SparkBulkInsertHelper.java
.../compact/HoodieSparkMergeOnReadTableCompactor.java
... and 98 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 039aeb6...c5471c3. Read the comment docs.

[wangxianghu]

This logic seems the same as LATEST ?

[veenaypatil]

@wangxianghu Yes, incase the offsets are not committed for consumer group we are setting it to latest, do you suggest to throw an exception instead ?

[wangxianghu]

@veenaypatil I mean in this feature, is there any difference between latest and none

copied from org.apache.kafka.clients.consumer.ConsumerConfig#AUTO_OFFSET_RESET_CONFIG

    /**
     * <code>auto.offset.reset</code>
     */
    public static final String AUTO_OFFSET_RESET_CONFIG = "auto.offset.reset";
    public static final String AUTO_OFFSET_RESET_DOC = "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): <ul><li>earliest: automatically reset the offset to the earliest offset<li>latest: automatically reset the offset to the latest offset</li><li>none: throw exception to the consumer if no previous offset is found for the consumer's group</li><li>anything else: throw exception to the consumer.</li></ul>";

[veenaypatil]

@wangxianghu

Yes, if you update the value to earliest/latest in this test case the consumer will start reading either from 0th offset (earliest) or 500th offset (latest), instead it should start from 250th offset (as this is the last committed offset)

hudi/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKafkaOffsetGen.java

Line 139 in 80a1f1c

KafkaOffsetGen kafkaOffsetGen = new KafkaOffsetGen(getConsumerConfigs("none"));

---

I actually don't like the NONE option here and wanted to use GROUP but the consumer will throw an exception in that case

[wangxianghu]

@wangxianghu

Yes, if you update the value to earliest/latest in this test case the consumer will start reading either from 0th offset (earliest) or 500th offset (latest), instead it should start from 250th offset (as this is the last committed offset)

hudi/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKafkaOffsetGen.java

Line 139 in 80a1f1c

KafkaOffsetGen kafkaOffsetGen = new KafkaOffsetGen(getConsumerConfigs("none"));

I actually don't like the NONE option here and wanted to use GROUP but the consumer will throw an exception in that case

IIUC, if you have committed the offset to kafka, when you start the consumer with the same group.id as before and set auto.offset.set to latest, it will continue to consume offset committed last time(which means 250th as you mentioned), right ?

[veenaypatil]

@wangxianghu yes, but that's not happening, I think it is because of the way we are explicitly setting the fromOffsets here - https://github.com/apache/hudi/blob/master/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java#L221 which moves the consumer to end, that is it starts reading from 500th offset for partition0 and partition1 in test case

Copying doc for KafkaConsumer#endOffsets

Get the end offsets for the given partitions. In the default read_uncommitted isolation level, the end offset is the high watermark (that is, the offset of the last successfully replicated message plus one)

[wangxianghu]

@wangxianghu yes, but that's not happening, I think it is because of the way we are explicitly setting the fromOffsets here - https://github.com/apache/hudi/blob/master/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java#L221 which moves the consumer to end, that is it starts reading from 500th offset for partition0 and partition1 in test case

Copying doc for KafkaConsumer#endOffsets

Get the end offsets for the given partitions. In the default read_uncommitted isolation level, the end offset is the high watermark (that is, the offset of the last successfully replicated message plus one)

you are right.

[wangxianghu]

currently, our org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen.KafkaResetOffsetStrategies is not the same as kafka, is it possible make them the same ?

[veenaypatil]

currently, our org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen.KafkaResetOffsetStrategies is not the same as kafka, is it possible make them the same ?

did not get you ? we have LATEST, EARLIEST and NONE which is same as Kafka auto reset config

[wangxianghu]

currently, our org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen.KafkaResetOffsetStrategies is not the same as kafka, is it possible make them the same ?

did not get you ? we have LATEST, EARLIEST and NONE which is same as Kafka auto reset config

hudi‘s LATEST means the end offsets for given partitions, no matter what the current offset is, while kafka's LATEST` will consumer from the last commit first. if there is no init offset it is the same as hudi's, right?

[veenaypatil]

@wangxianghu IMO, we should not update the existing functionality, as this can impact the existing Hudi users, till now Hudi did not have a provision to read from committed offset, this just adds an option for the users to set auto.offset.reset to none.

We can think if we should take the option as auto.offset.reset=group from user and then set to none in code so that the consumer does not throw exception.
This change aligns with the way Flink also provides various options to the user - https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/datastream/kafka/#kafka-consumers-start-position-configuration

[wangxianghu]

@wangxianghu IMO, we should not update the existing functionality, as this can impact the existing Hudi users, till now Hudi did not have a provision to read from committed offset, this just adds an option for the users to set auto.offset.reset to none.

We can think if we should take the option as auto.offset.reset=group from user and then set to none in code so that the consumer does not throw exception.
This change aligns with the way Flink also provides various options to the user - https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/datastream/kafka/#kafka-consumers-start-position-configuration

I agree with you.
I think you can give a try

[veenaypatil]

@wangxianghu as per our discussion, changes are done, pls review

[wangxianghu]

@veenaypatil thanks for your patience, LGTM now
will merge when the ci is green