Add UnboundedReaderMaxReadTimeMs to DataflowPipelineDebugOptions, deprecate UnboundedReaderMaxReadTimeSec #31091
Conversation
|
related to #31082 |
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
|
Run Java Unit Tests |
|
R: @jrmccluskey Jack, would you be able to take a look at this or suggest reviewer? |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control |
...ava/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineDebugOptions.java
Show resolved
Hide resolved
| * The max amount of time an UnboundedReader is consumed before checkpointing. | ||
| * | ||
| * @deprecated use {@link DataflowPipelineDebugOptions#getUnboundedReaderMaxReadTimeMs()} instead | ||
| */ | ||
| @Description( | ||
| "The max amount of time before an UnboundedReader is consumed before checkpointing, in seconds.") | ||
| @Default.Integer(10) |
There was a problem hiding this comment.
can the default be removed here? I think it will be less brittle since the default won't be repeated in the WorkerCustomSources
if (maxReadTimeMsDeprecated != null and maxReadTimeMsDeprecated > 0) {
// use it. Seems fine just pick if both were set.
} else {
use new option, either the default or explicitly set
}
or another option could be to use a DefaultFactory (see StagerFactory in this file) so that the default of the new field is to be 1000 * getUnboundedReaderMaxReadTimeSec
Then the code could just use getUnboundedReaderMaxReadTimeMs() and it would either default to fallback to previous value (set or default) or prefer the new value if it was set.
...ava/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineDebugOptions.java
Show resolved
Hide resolved
| * Sets Integer value based on old, deprecated field ({@link | ||
| * DataflowPipelineDebugOptions#getUnboundedReaderMaxReadTimeSec()}) is set. | ||
| */ | ||
| class UnboundedReaderMaxReadTimeFactory implements DefaultValueFactory<Integer> { |
There was a problem hiding this comment.
can this be static final class?
|
|
||
| /** | ||
| * Sets Integer value based on old, deprecated field ({@link | ||
| * DataflowPipelineDebugOptions#getUnboundedReaderMaxReadTimeSec()}) is set. |
UnboundedReader is controlled with UnboundedReaderMaxReadTimeSec and UnboundedReaderMaxElements.
For sub second latency user has to set UnboundedReaderMaxElements to 1 which is not flexible as there may be multiple situation where there are more elements.
In this change UnboundedReaderMaxReadTimeSec becomes deprecated.
UnboundedReaderMaxReadTimeMs is introduced.
Change is backward compatible - it takes into account both values - If UnboundedReaderMaxReadTimeSec is non default it will take overwrite UnboundedReaderMaxReadTimeMs.