lincoln-lil commented on code in PR #29005:
URL: https://github.com/apache/flink/pull/29005#discussion_r3840574130
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java:
##########
@@ -522,7 +528,9 @@ private void timerTriggered() throws Exception {
public void complete(Collection<OUT> results) {
Preconditions.checkNotNull(
results, "Results must not be null, use empty collection
to emit nothing");
- if (!retryDisabledOnFinish.get() &&
resultHandler.inputRecord.isRecord()) {
+ if (!timedOut.get()
+ && !retryDisabledOnFinish.get()
+ && resultHandler.inputRecord.isRecord()) {
Review Comment:
nit: Could we extract this repeated condition into a helper such as
`shouldProcessResultForRetry()`? The same retry-routing decision is used by all
three `complete*` overloads.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorTest.java:
##########
@@ -1399,6 +1399,69 @@ void
testProcessingTimeWithTimeoutFunctionUnorderedWithRetry() throws Exception
testProcessingTimeAlwaysTimeoutFunctionWithRetry(AsyncDataStream.OutputMode.UNORDERED);
}
+ // timeout+retry with serialized async completions must still emit every
timeout result
+ @Test
+ void reproSerializedCompletionsWithRetry() throws Exception {
+ StreamTaskMailboxTestHarnessBuilder<Integer> builder =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ OneInputStreamTask::new,
BasicTypeInfo.INT_TYPE_INFO)
+ .addInput(BasicTypeInfo.INT_TYPE_INFO);
+
+ AsyncRetryStrategy exceptionRetryStrategy =
+ new AsyncRetryStrategies.FixedDelayRetryStrategyBuilder(5,
100L)
+ .ifException(RetryPredicates.HAS_EXCEPTION_PREDICATE)
+ .build();
+
+ try (StreamTaskMailboxTestHarness<Integer> testHarness =
+ builder.setupOutputForSingletonOperatorChain(
+ new AsyncWaitOperatorFactory<>(
+ new
AlwaysTimeoutSingleThreadAsyncFunction(),
+ TIMEOUT,
+ 10,
+ AsyncDataStream.OutputMode.UNORDERED,
+ exceptionRetryStrategy))
+ .build()) {
+
+ testHarness.processElement(new StreamRecord<>(1, 1L));
+ testHarness.processElement(new StreamRecord<>(2, 2L));
+
+ long deadlineNanos = System.nanoTime() +
TimeUnit.SECONDS.toNanos(5);
+ while (testHarness.getOutput().size() < 2) {
+ testHarness.processAll();
+ if (System.nanoTime() > deadlineNanos) {
+ throw new AssertionError(
+ "REPRO CONFIRMED: operator produced only "
+ + testHarness.getOutput().size()
+ + "/2 outputs within 5s when async
completions are serialized "
+ + "(single-thread executor). Timeout
default value was dropped.");
+ }
+ //noinspection BusyWait
+ Thread.sleep(50);
+ }
+ }
+ }
+
+ private static class AlwaysTimeoutSingleThreadAsyncFunction
+ extends AlwaysTimeoutWithDefaultValueAsyncFunction {
+ private static final long serialVersionUID = 2L;
+ private static final ExecutorService POOL =
Executors.newSingleThreadExecutor();
+
+ @Override
+ public void asyncInvoke(Integer input, ResultFuture<Integer>
resultFuture) {
+ tryCounts.merge(input, 1, Integer::sum);
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ Thread.sleep(501L);
Review Comment:
It might be safer to coordinate this race with latches or manual time, as
wall-clock sleeps may not always exercise the intended retry/timeout
interleaving.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorTest.java:
##########
@@ -1399,6 +1399,69 @@ void
testProcessingTimeWithTimeoutFunctionUnorderedWithRetry() throws Exception
testProcessingTimeAlwaysTimeoutFunctionWithRetry(AsyncDataStream.OutputMode.UNORDERED);
}
+ // timeout+retry with serialized async completions must still emit every
timeout result
+ @Test
+ void reproSerializedCompletionsWithRetry() throws Exception {
+ StreamTaskMailboxTestHarnessBuilder<Integer> builder =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ OneInputStreamTask::new,
BasicTypeInfo.INT_TYPE_INFO)
+ .addInput(BasicTypeInfo.INT_TYPE_INFO);
+
+ AsyncRetryStrategy exceptionRetryStrategy =
+ new AsyncRetryStrategies.FixedDelayRetryStrategyBuilder(5,
100L)
+ .ifException(RetryPredicates.HAS_EXCEPTION_PREDICATE)
+ .build();
+
+ try (StreamTaskMailboxTestHarness<Integer> testHarness =
+ builder.setupOutputForSingletonOperatorChain(
+ new AsyncWaitOperatorFactory<>(
+ new
AlwaysTimeoutSingleThreadAsyncFunction(),
+ TIMEOUT,
+ 10,
+ AsyncDataStream.OutputMode.UNORDERED,
+ exceptionRetryStrategy))
+ .build()) {
+
+ testHarness.processElement(new StreamRecord<>(1, 1L));
+ testHarness.processElement(new StreamRecord<>(2, 2L));
+
+ long deadlineNanos = System.nanoTime() +
TimeUnit.SECONDS.toNanos(5);
+ while (testHarness.getOutput().size() < 2) {
+ testHarness.processAll();
+ if (System.nanoTime() > deadlineNanos) {
+ throw new AssertionError(
+ "REPRO CONFIRMED: operator produced only "
+ + testHarness.getOutput().size()
+ + "/2 outputs within 5s when async
completions are serialized "
+ + "(single-thread executor). Timeout
default value was dropped.");
+ }
+ //noinspection BusyWait
+ Thread.sleep(50);
+ }
+ }
+ }
+
+ private static class AlwaysTimeoutSingleThreadAsyncFunction
+ extends AlwaysTimeoutWithDefaultValueAsyncFunction {
+ private static final long serialVersionUID = 2L;
+ private static final ExecutorService POOL =
Executors.newSingleThreadExecutor();
Review Comment:
It would be good to clean up this executor after the test so that its
non-daemon thread and pending callbacks do not outlive the test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
