[FLINK-35299] Respect initial position for new streams #140
Conversation
…reams - disabled by default so that this can go in a minor release. Default can be adjusted for a major.
|
Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html) |
antssilva96
changed the title
antssilva96
changed the title
| public static final String APPLY_STREAM_INITIAL_POSITION_FOR_NEW_STREAMS = | ||
| "flink.stream.initpos-for-new-streams"; | ||
|
|
||
| /** | ||
| * Property that can be used to ignore the restore state for a particular stream and instead use | ||
| * the initial position. This is useful to reset a specific stream to consume from TRIM_HORIZON | ||
| * or LATEST if needed. Values must be passed in a comma separated list. | ||
| * | ||
| * <p>If a stream is in this list, it will use initial position regardless of the value of the | ||
| * {@link #APPLY_STREAM_INITIAL_POSITION_FOR_NEW_STREAMS} property. | ||
| */ | ||
| public static final String STREAMS_TO_APPLY_STREAM_INITIAL_POSITION_TO = | ||
| "flink.stream.initpos-streams"; | ||
|
|
There was a problem hiding this comment.
not really sure about these property names... any suggestions ?
|
|
||
| For example, if you configure your application with | ||
| ``` | ||
| consumerConfig.put(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "LATEST"); |
There was a problem hiding this comment.
I was wondering if we are able to set different streams with different INITIAL POSITION. Let's say we would add streamA, streamB and streamC as new streams, I want to have streamA and streamB to consume from LATEST and streamC from AT_TIMESTAMP. Is this possible?
There was a problem hiding this comment.
This was not possible before and is still not possible now 😅
Even though it would be possible to do that with minimal changes, I feel like that is another feature request on its own and probably doesn't belong in this one.
The main goal of this change is: allow users to say when INITIAL position should be used regardless of the stored state.
Purpose of the change
According to the javadoc, the
STREAM_INITIAL_POSITIONproperty defines where to start reading Kinesis streams from. However, in the current implementation this is only true if there isn't any restore state at all for any streams for that KinesisConsumer, otherwise the new stream is handled the same way a new shard for and existing stream is: start consuming from EARLIEST (same as TRIM_HORIZON initial position).This MR changes that by making
FlinkKinesisConsumerto useSTREAM_INITIAL_POSITIONconfig for new streams, which aligns with that is documented.This behavior is disabled by default to not introduce a breaking change, but can be enabled by setting
flink.stream.initpos-for-new-streamsto true.Additionally, a second config was created -
flink.stream.initpos-streams- to allow specific streams to be "reset" to whatever the STREAM_INITIAL_POSITION is defined. This is an important addition in this MR because users who notice this bug and want to enable the correct behaviour will want to reset the now recorded offset for the new stream.Verifying this change
TODO
Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing
(Please pick either of the following options)
This change is a trivial rework / code cleanup without any test coverage.
(or)
This change is already covered by existing tests, such as (please describe tests).
(or)
This change added tests and can be verified as follows:
(example:)
Significant changes
(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for convenience.)
@Public(Evolving))