[Pull-based Ingestion] Offset management, support rewind by offset or timestamp

[xuxiong1]

Description

This PR supports starting the ingestion from a user-specified timestamp or offset when creating the new index.

• added two new ResetState: REWIND_BY_OFFSET and REWIND_BY_TIMESTAMP
• added new setting ingestion_source.pointer.init.reset.value to IndexMetadata, controls the rewind offset or timestamp in millis
• added corresponding integration tests

Related Issues

Resolves #17318

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

❌ Gradle check result for 7c573c3: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[Bukhtawar]

As I understand the index level settings are controlling the offset resets. I am assuming that we expect this to work in a single shard and single partition setup

[github-actions]

❌ Gradle check result for 4a83c12: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 54c8bbe: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 9dd5bd0: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[yupeng9]

Looks good to me

[github-actions]

❌ Gradle check result for 9dd5bd0: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for c76a342: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for c76a342: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 3d35f46: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

✅ Gradle check result for 3d35f46: SUCCESS

[codecov]

Codecov Report

Attention: Patch coverage is 52.05479% with 35 lines in your changes missing coverage. Please review.

Project coverage is 72.46%. Comparing base (8447737) to head (3d35f46).
Report is 5 commits behind head on main.

Files with missing lines	Patch %	Lines
...pensearch/plugin/kafka/KafkaPartitionConsumer.java	28.00%	14 Missing and 4 partials ⚠️
...org/opensearch/cluster/metadata/IndexMetadata.java	47.36%	5 Missing and 5 partials ⚠️
...g/opensearch/cluster/metadata/IngestionSource.java	69.23%	4 Missing ⚠️
...rch/indices/pollingingest/DefaultStreamPoller.java	57.14%	3 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #17354      +/-   ##
============================================
- Coverage     72.48%   72.46%   -0.03%     
- Complexity    65637    65672      +35     
============================================
  Files          5303     5303              
  Lines        304793   304858      +65     
  Branches      44202    44217      +15     
============================================
- Hits         220926   220907      -19     
- Misses        65711    65794      +83     
- Partials      18156    18157       +1     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.