Skip to content

Change type for UnboundedReaderMaxReadTimeSec #31037

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jrmccluskey merged 11 commits into from
Apr 18, 2024
Merged

Change type for UnboundedReaderMaxReadTimeSec #31037

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
c9ac05b
add ms part for UnboundedReader checkpointing
stankiewicz Apr 18, 2024
a36f178
typo
stankiewicz Apr 18, 2024
cf5192a
spotless
stankiewicz Apr 18, 2024
a001d69
spotless
stankiewicz Apr 18, 2024
dff48be
spotless
stankiewicz Apr 18, 2024
8bb39f9
[IntLongMath] Expression of type int may overflow before being assign…
stankiewicz Apr 18, 2024
852efeb
readerMaxReadTime sec as double
stankiewicz Apr 18, 2024
8fc8062
readerMaxReadTime sec as double
stankiewicz Apr 18, 2024
87fca4c
readerMaxReadTime sec as double
stankiewicz Apr 18, 2024
5ec8f0b
readerMaxReadTime sec as double
stankiewicz Apr 18, 2024
445d970
spotless
stankiewicz Apr 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions .../src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineDebugOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -218,11 +218,12 @@ public Dataflow create(PipelineOptions options) {

/** The max amount of time an UnboundedReader is consumed before checkpointing. */
@Description(
Copy link
Contributor

@jrmccluskey jrmccluskey Apr 18, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Mentioning that the representation can be in fractions of seconds is probably a good call-out here

Copy link
Contributor Author

@stankiewicz stankiewicz Apr 18, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point. added additional comment.

"The max amount of time before an UnboundedReader is consumed before checkpointing, in seconds.")
@Default.Integer(10)
Integer getUnboundedReaderMaxReadTimeSec();
"The max amount of time before an UnboundedReader is consumed before checkpointing, "
+ "in seconds. Duration can be set to fractions of seconds. ")
@Default.Double(10.0)
double getUnboundedReaderMaxReadTimeSec();

void setUnboundedReaderMaxReadTimeSec(Integer value);
void setUnboundedReaderMaxReadTimeSec(double value);

/** The max elements read from an UnboundedReader before checkpointing. */
@Description("The max elements read from an UnboundedReader before checkpointing. ")
Expand Down
4 changes: 3 additions & 1 deletion ...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WorkerCustomSources.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -798,7 +798,9 @@ private UnboundedReaderIterator(
DataflowPipelineDebugOptions debugOptions = options.as(DataflowPipelineDebugOptions.class);
this.endTime =
Instant.now()
.plus(Duration.standardSeconds(debugOptions.getUnboundedReaderMaxReadTimeSec()));
.plus(
Duration.millis(
(long) (debugOptions.getUnboundedReaderMaxReadTimeSec() * 1000L)));
this.maxElems = debugOptions.getUnboundedReaderMaxElements();
this.backoffFactory =
FluentBackoff.DEFAULT
Expand Down
7 changes: 4 additions & 3 deletions ...worker/src/test/java/org/apache/beam/runners/dataflow/worker/WorkerCustomSourcesTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -598,6 +598,7 @@ public void testReadUnboundedReader() throws Exception {
int maxElements = 10;
DataflowPipelineDebugOptions debugOptions = options.as(DataflowPipelineDebugOptions.class);
debugOptions.setUnboundedReaderMaxElements(maxElements);
debugOptions.setUnboundedReaderMaxReadTimeSec(10);

ByteString state = ByteString.EMPTY;
for (int i = 0; i < 10 * maxElements;
Expand Down Expand Up @@ -645,10 +646,10 @@ public void testReadUnboundedReader() throws Exception {
numReadOnThisIteration++;
}
Instant afterReading = Instant.now();
long maxReadSec = debugOptions.getUnboundedReaderMaxReadTimeSec();
double maxReadSec = debugOptions.getUnboundedReaderMaxReadTimeSec();
assertThat(
new Duration(beforeReading, afterReading).getStandardSeconds(),
lessThanOrEqualTo(maxReadSec + 1));
new Duration(beforeReading, afterReading).getMillis(),
lessThanOrEqualTo((long) ((maxReadSec + 1) * 1000L)));
assertThat(
numReadOnThisIteration, lessThanOrEqualTo(debugOptions.getUnboundedReaderMaxElements()));

Expand Down