[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

[yupeng9]

we can fallback to the policy of auto.offset.reset?

[xuxiong1]

Usually, the message not found could happen when the user specifies a timestamp larger than the latest in topic, in this case, return to the latest offset makes sense.
If the user specifies a timestamp smaller than the earliest, the offset will be automatically set to 0 as the earliest.

[yupeng9]

is this dynamic?

[xuxiong1]

actually i followed the same as the index.ingestion_source.pointer.init.reset, which is also set to dynamic.
I'll change both of them to final.

[yupeng9]

how many types do we have? shall we consider enum?

[xuxiong1]

right, it will be mapped to enum StreamPoller.ResetState

    /**
     *  a reset state to indicate how to reset the pointer
     */
    @ExperimentalApi
    enum ResetState {
        EARLIEST,
        LATEST,
        REWIND_BY_OFFSET,
        REWIND_BY_TIMESTAMP,
        NONE,
    }

I'll replace it with ResetState enum here to be more consistent.

[yupeng9]

this challenge is that not all streaming sources use long for offset, for example, kinesis uses 128 bit.
we can use reset from a given pointer

[xuxiong1]

I see, the problem is we're passing a long value from setting.

    public static final Setting<Long> INGESTION_SOURCE_POINTER_INIT_RESET_VALUE_SETTING = Setting.longSetting(
        SETTING_INGESTION_SOURCE_POINTER_INIT_RESET_VALUE,
        0,
        0,
        Property.IndexScope,
        Property.Final
    );

in that case, passing a long value might not be sufficient. I'll try to pass a string value.

[yupeng9]

again reset value cannot always be represented in long

[xuxiong1]

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

should also work for multi-shard setup, shards will be mapped to partitions.

[github-actions]

❌ Gradle check result for 21eefdd: 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 701a54f: 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?