This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 0b75ad105ba [FLINK-37243] Adds main thread callback to
AsyncWaitOperator (#26122)
0b75ad105ba is described below
commit 0b75ad105baa466bb235c93406993352b2dad791
Author: Alan Sheinberg <[email protected]>
AuthorDate: Tue Mar 18 09:05:39 2025 -0700
[FLINK-37243] Adds main thread callback to AsyncWaitOperator (#26122)
[FLINK-37243] Adds main thread callback to AsyncWaitOperator
Also now uses this callback in DelegatingAsyncResultFuture to do the
converter call on the main thread.
---
.../{ResultFuture.java => CollectionSupplier.java} | 24 +---
.../api/functions/async/ResultFuture.java | 9 ++
.../api/operators/async/AsyncWaitOperator.java | 37 +++++
.../async/queue/StreamElementQueueEntry.java | 5 +
.../api/operators/async/AsyncWaitOperatorTest.java | 159 +++++++++++++++++++++
.../tasks/StreamTaskMailboxTestHarness.java | 4 +
.../planner/codegen/AsyncCodeGeneratorTest.java | 10 ++
.../collector/TableFunctionResultFuture.java | 10 ++
.../calc/async/DelegatingAsyncResultFuture.java | 16 ++-
.../join/lookup/AsyncLookupJoinRunner.java | 19 +++
10 files changed, 269 insertions(+), 24 deletions(-)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/ResultFuture.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/CollectionSupplier.java
similarity index 57%
copy from
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/ResultFuture.java
copy to
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/CollectionSupplier.java
index d52575b5fd9..a45457eeab6 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/ResultFuture.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/CollectionSupplier.java
@@ -23,28 +23,16 @@ import org.apache.flink.annotation.PublicEvolving;
import java.util.Collection;
/**
- * {@link ResultFuture} collects data / error in user codes while processing
async i/o.
+ * Supplies a collection of output. Used in {@link ResultFuture} to provide a
supplier for results.
*
- * @param <OUT> Output type
+ * @param <OUT> The type of the output.
*/
@PublicEvolving
-public interface ResultFuture<OUT> {
+public interface CollectionSupplier<OUT> {
/**
- * Completes the result future with a collection of result objects.
+ * Returns the collection of results.
*
- * <p>Note that it should be called for exactly one time in the user code.
Calling this function
- * for multiple times will cause data lose.
- *
- * <p>Put all results in a {@link Collection} and then emit output.
- *
- * @param result A list of results.
- */
- void complete(Collection<OUT> result);
-
- /**
- * Completes the result future exceptionally with an exception.
- *
- * @param error A Throwable object.
+ * @throws Exception
*/
- void completeExceptionally(Throwable error);
+ Collection<OUT> get() throws Exception;
}
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/ResultFuture.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/ResultFuture.java
index d52575b5fd9..7fcf44293a7 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/ResultFuture.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/ResultFuture.java
@@ -47,4 +47,13 @@ public interface ResultFuture<OUT> {
* @param error A Throwable object.
*/
void completeExceptionally(Throwable error);
+
+ /**
+ * The same as complete, but will execute the supplier on the Mailbox
thread which initiated the
+ * asynchronous process.
+ *
+ * <p>Note that if an exception is thrown while executing the supplier,
the result should be the
+ * same as calling {@link ResultFuture#completeExceptionally(Throwable)}.
+ */
+ void complete(CollectionSupplier<OUT> supplier);
}
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
index 5e7cf8e8a8b..af418de2df2 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
@@ -29,6 +29,7 @@ import
org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.AsyncDataStream.OutputMode;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.AsyncRetryStrategy;
+import org.apache.flink.streaming.api.functions.async.CollectionSupplier;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
@@ -505,6 +506,27 @@ public class AsyncWaitOperator<IN, OUT>
}
}
+ @Override
+ public void complete(CollectionSupplier<OUT> supplier) {
+ Preconditions.checkNotNull(
+ supplier, "Runnable must not be null, return empty
collection to emit nothing");
+ if (!retryDisabledOnFinish.get() &&
resultHandler.inputRecord.isRecord()) {
+ mailboxExecutor.submit(
+ () -> {
+ try {
+ processRetry(supplier.get(), null);
+ } catch (Throwable t) {
+ processRetry(null, t);
+ }
+ },
+ "RetryableResultHandlerDelegator#complete");
+ } else {
+ cancelRetryTimer();
+
+ resultHandler.complete(supplier);
+ }
+ }
+
private void processRetryInMailBox(Collection<OUT> results, Throwable
error) {
mailboxExecutor.execute(
() -> processRetry(results, error), "delayed retry or
complete");
@@ -600,6 +622,21 @@ public class AsyncWaitOperator<IN, OUT>
processInMailbox(results);
}
+ @Override
+ public void complete(CollectionSupplier<OUT> supplier) {
+ // already completed (exceptionally or with previous complete call
from ill-written
+ // AsyncFunction), so ignore additional result
+ if (!completed.compareAndSet(false, true)) {
+ return;
+ }
+ mailboxExecutor.execute(
+ () -> {
+ // If there is an exception, let it bubble up and fail
the job.
+ processResults(supplier.get());
+ },
+ "ResultHandler#complete");
+ }
+
private void processInMailbox(Collection<OUT> results) {
// move further processing into the mailbox thread
mailboxExecutor.execute(
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/StreamElementQueueEntry.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/StreamElementQueueEntry.java
index 4ff960036cd..fdeb3f20701 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/StreamElementQueueEntry.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/StreamElementQueueEntry.java
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.api.operators.async.queue;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.streaming.api.functions.async.CollectionSupplier;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
@@ -62,4 +63,8 @@ interface StreamElementQueueEntry<OUT> extends
ResultFuture<OUT> {
throw new UnsupportedOperationException(
"This result future should only be used to set completed
results.");
}
+
+ default void complete(CollectionSupplier<OUT> supplier) {
+ throw new UnsupportedOperationException();
+ }
}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorTest.java
index 500f86d62da..b0a33a0cf48 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorTest.java
@@ -35,6 +35,7 @@ import
org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.operators.testutils.ExpectedTestException;
import org.apache.flink.runtime.operators.testutils.MockEnvironment;
import org.apache.flink.runtime.state.TestTaskStateManager;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
@@ -69,6 +70,8 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.RegisterExtension;
+import javax.annotation.Nullable;
+
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
@@ -90,9 +93,12 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
+import static
org.apache.flink.streaming.util.retryable.AsyncRetryStrategies.NO_RETRY_STRATEGY;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertEquals;
/**
* Tests for {@link AsyncWaitOperator}. These test that:
@@ -1324,6 +1330,125 @@ class AsyncWaitOperatorTest {
}
}
+ @Test
+ public void testProcessingTimeWithMailboxThreadOrdered() throws Exception {
+ testProcessingTimeWithCollectFromMailboxThread(
+ AsyncDataStream.OutputMode.ORDERED, NO_RETRY_STRATEGY);
+ }
+
+ @Test
+ public void testProcessingTimeWithMailboxThreadUnordered() throws
Exception {
+ testProcessingTimeWithCollectFromMailboxThread(
+ AsyncDataStream.OutputMode.UNORDERED, NO_RETRY_STRATEGY);
+ }
+
+ @Test
+ public void testProcessingTimeWithMailboxThreadOrderedWithRetry() throws
Exception {
+ testProcessingTimeWithCollectFromMailboxThread(
+ AsyncDataStream.OutputMode.ORDERED, exceptionRetryStrategy);
+ }
+
+ @Test
+ public void testProcessingTimeWithMailboxThreadUnorderedWithRetry() throws
Exception {
+ testProcessingTimeWithCollectFromMailboxThread(
+ AsyncDataStream.OutputMode.UNORDERED, exceptionRetryStrategy);
+ }
+
+ @Test
+ public void testProcessingTimeWithErrorFromMailboxThread() throws
Exception {
+ testProcessingTimeWithErrorFromMailboxThread(NO_RETRY_STRATEGY);
+ }
+
+ @Test
+ public void testProcessingTimeWithMailboxThreadErrorWithRetry() throws
Exception {
+ testProcessingTimeWithErrorFromMailboxThread(exceptionRetryStrategy);
+ }
+
+ private void testProcessingTimeWithErrorFromMailboxThread(
+ @Nullable AsyncRetryStrategy<Integer> asyncRetryStrategy) throws
Exception {
+ StreamTaskMailboxTestHarnessBuilder<Integer> builder =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ OneInputStreamTask::new,
BasicTypeInfo.INT_TYPE_INFO)
+ .addInput(BasicTypeInfo.INT_TYPE_INFO);
+ try (StreamTaskMailboxTestHarness<Integer> testHarness =
+ builder.setupOutputForSingletonOperatorChain(
+ new AsyncWaitOperatorFactory<>(
+ new CallThreadAsyncFunctionError(),
+ TIMEOUT,
+ 1,
+ AsyncDataStream.OutputMode.UNORDERED,
+ asyncRetryStrategy))
+ .build()) {
+ final long initialTime = 0L;
+ AtomicReference<Throwable> error = new AtomicReference<>();
+
testHarness.getStreamMockEnvironment().setExternalExceptionHandler(error::set);
+
+ // Sometimes, processElement invoke the async function
immediately, so we should catch
+ // any exception.
+ try {
+ testHarness.processElement(new StreamRecord<>(1, initialTime +
1));
+ while (error.get() == null) {
+ testHarness.processAll();
+ }
+ } catch (Exception e) {
+ // This simulates a mailbox failure failing the job
+ error.set(e);
+ }
+
+ ExceptionUtils.assertThrowable(error.get(),
ExpectedTestException.class);
+
+ testHarness.endInput();
+ }
+ }
+
+ private void testProcessingTimeWithCollectFromMailboxThread(
+ AsyncDataStream.OutputMode mode,
+ @Nullable AsyncRetryStrategy<Integer> asyncRetryStrategy)
+ throws Exception {
+ StreamTaskMailboxTestHarnessBuilder<Integer> builder =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ OneInputStreamTask::new,
BasicTypeInfo.INT_TYPE_INFO)
+ .addInput(BasicTypeInfo.INT_TYPE_INFO);
+ try (StreamTaskMailboxTestHarness<Integer> testHarness =
+ builder.setupOutputForSingletonOperatorChain(
+ new AsyncWaitOperatorFactory<>(
+ new CallThreadAsyncFunction(),
+ TIMEOUT,
+ 3,
+ mode,
+ asyncRetryStrategy))
+ .build()) {
+
+ final long initialTime = 0L;
+ final Queue<Object> expectedOutput = new ArrayDeque<>();
+
+ testHarness.processElement(new StreamRecord<>(1, initialTime + 1));
+ testHarness.processElement(new StreamRecord<>(2, initialTime + 2));
+ testHarness.processElement(new StreamRecord<>(3, initialTime + 3));
+
+ expectedOutput.add(new StreamRecord<>(2, initialTime + 1));
+ expectedOutput.add(new StreamRecord<>(4, initialTime + 2));
+ expectedOutput.add(new StreamRecord<>(6, initialTime + 3));
+
+ while (testHarness.getOutput().size() < expectedOutput.size()) {
+ testHarness.processAll();
+ }
+
+ if (mode == AsyncDataStream.OutputMode.ORDERED) {
+ TestHarnessUtil.assertOutputEquals(
+ "ORDERED Output was not correct.", expectedOutput,
testHarness.getOutput());
+ } else {
+ TestHarnessUtil.assertOutputEqualsSorted(
+ "UNORDERED Output was not correct.",
+ expectedOutput,
+ testHarness.getOutput(),
+ new StreamRecordComparator());
+ }
+
+ testHarness.endInput();
+ }
+ }
+
private static class CollectableFuturesAsyncFunction<IN> implements
AsyncFunction<IN, IN> {
private static final long serialVersionUID = -4214078239227288637L;
@@ -1431,4 +1556,38 @@ class AsyncWaitOperatorTest {
resultFuture.complete(Collections.singletonList(-1));
}
}
+
+ private static class CallThreadAsyncFunction extends
MyAbstractAsyncFunction<Integer> {
+ private static final long serialVersionUID = -1504699677704123889L;
+
+ @Override
+ public void asyncInvoke(final Integer input, final
ResultFuture<Integer> resultFuture)
+ throws Exception {
+ final Thread callThread = Thread.currentThread();
+ executorService.submit(
+ () ->
+ resultFuture.complete(
+ () -> {
+ assertEquals(callThread,
Thread.currentThread());
+ return Collections.singletonList(input
* 2);
+ }));
+ }
+ }
+
+ private static class CallThreadAsyncFunctionError extends
MyAbstractAsyncFunction<Integer> {
+ private static final long serialVersionUID = -1504699677704123889L;
+
+ @Override
+ public void asyncInvoke(final Integer input, final
ResultFuture<Integer> resultFuture)
+ throws Exception {
+ Thread callThread = Thread.currentThread();
+ executorService.submit(
+ () ->
+ resultFuture.complete(
+ () -> {
+ assertEquals(callThread,
Thread.currentThread());
+ throw new ExpectedTestException();
+ }));
+ }
+ }
}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarness.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarness.java
index 4e78f4db011..ddd06e313e3 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarness.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarness.java
@@ -195,4 +195,8 @@ public class StreamTaskMailboxTestHarness<OUT> implements
AutoCloseable {
public TestCheckpointResponder getCheckpointResponder() {
return (TestCheckpointResponder)
taskStateManager.getCheckpointResponder();
}
+
+ public StreamMockEnvironment getStreamMockEnvironment() {
+ return streamMockEnvironment;
+ }
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/AsyncCodeGeneratorTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/AsyncCodeGeneratorTest.java
index 79f867c2843..d5584de18b6 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/AsyncCodeGeneratorTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/AsyncCodeGeneratorTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.planner.codegen;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
+import org.apache.flink.streaming.api.functions.async.CollectionSupplier;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
@@ -198,5 +199,14 @@ public class AsyncCodeGeneratorTest {
public CompletableFuture<Collection<RowData>> getResult() {
return data;
}
+
+ @Override
+ public void complete(CollectionSupplier<RowData> supplier) {
+ try {
+ data.complete(supplier.get());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
}
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/collector/TableFunctionResultFuture.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/collector/TableFunctionResultFuture.java
index 28c9da1697c..559e8a36e9d 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/collector/TableFunctionResultFuture.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/collector/TableFunctionResultFuture.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.runtime.collector;
import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.streaming.api.functions.async.CollectionSupplier;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
/** The basic implementation of collector for {@link ResultFuture} in table
joining. */
@@ -60,4 +61,13 @@ public abstract class TableFunctionResultFuture<T> extends
AbstractRichFunction
public void completeExceptionally(Throwable error) {
this.resultFuture.completeExceptionally(error);
}
+
+ /**
+ * Unsupported, because the containing classes are AsyncFunctions which
don't have access to the
+ * mailbox to invoke from the caller thread.
+ */
+ @Override
+ public void complete(CollectionSupplier<T> supplier) {
+ throw new UnsupportedOperationException();
+ }
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/DelegatingAsyncResultFuture.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/DelegatingAsyncResultFuture.java
index 2a42a3414bd..dd6ae830ce0 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/DelegatingAsyncResultFuture.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/DelegatingAsyncResultFuture.java
@@ -24,6 +24,7 @@ import
org.apache.flink.table.data.conversion.DataStructureConverter;
import org.apache.flink.util.Preconditions;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.function.BiConsumer;
@@ -39,7 +40,7 @@ public class DelegatingAsyncResultFuture implements
BiConsumer<Object, Throwable
private final List<Object> synchronousResults = new ArrayList<>();
private Function<Object, RowData> outputFactory;
private CompletableFuture<Object> future;
- private CompletableFuture<Object> convertedFuture;
+ private DataStructureConverter<Object, Object> converter;
public DelegatingAsyncResultFuture(ResultFuture<Object>
delegatedResultFuture) {
this.delegatedResultFuture = delegatedResultFuture;
@@ -60,11 +61,11 @@ public class DelegatingAsyncResultFuture implements
BiConsumer<Object, Throwable
public CompletableFuture<?> createAsyncFuture(
DataStructureConverter<Object, Object> converter) {
Preconditions.checkState(future == null);
- Preconditions.checkState(convertedFuture == null);
+ Preconditions.checkState(this.converter == null);
Preconditions.checkNotNull(outputFactory);
future = new CompletableFuture<>();
- convertedFuture = future.thenApply(converter::toInternal);
- this.convertedFuture.whenComplete(this);
+ this.converter = converter;
+ future.whenComplete(this);
return future;
}
@@ -74,8 +75,11 @@ public class DelegatingAsyncResultFuture implements
BiConsumer<Object, Throwable
delegatedResultFuture.completeExceptionally(throwable);
} else {
try {
- RowData rowData = outputFactory.apply(o);
-
delegatedResultFuture.complete(java.util.Collections.singleton(rowData));
+ delegatedResultFuture.complete(
+ () -> {
+ Object converted = converter.toInternal(o);
+ return
Collections.singleton(outputFactory.apply(converted));
+ });
} catch (Throwable t) {
delegatedResultFuture.completeExceptionally(t);
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/lookup/AsyncLookupJoinRunner.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/lookup/AsyncLookupJoinRunner.java
index 30c96231dac..73f9575971b 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/lookup/AsyncLookupJoinRunner.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/lookup/AsyncLookupJoinRunner.java
@@ -24,6 +24,7 @@ import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.functions.util.FunctionUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
+import org.apache.flink.streaming.api.functions.async.CollectionSupplier;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.table.data.GenericRowData;
@@ -274,6 +275,15 @@ public class AsyncLookupJoinRunner extends
RichAsyncFunction<RowData, RowData> {
realOutput.completeExceptionally(error);
}
+ /**
+ * Unsupported, because the containing classes are AsyncFunctions
which don't have access to
+ * the mailbox to invoke from the caller thread.
+ */
+ @Override
+ public void complete(CollectionSupplier<Object> supplier) {
+ throw new UnsupportedOperationException();
+ }
+
public void close() throws Exception {
joinConditionResultFuture.close();
}
@@ -295,6 +305,15 @@ public class AsyncLookupJoinRunner extends
RichAsyncFunction<RowData, RowData> {
public void completeExceptionally(Throwable error) {
JoinedRowResultFuture.this.completeExceptionally(error);
}
+
+ /**
+ * Unsupported, because the containing classes are AsyncFunctions
which don't have
+ * access to the mailbox to invoke from the caller thread.
+ */
+ @Override
+ public void complete(CollectionSupplier<RowData> supplier) {
+ throw new UnsupportedOperationException();
+ }
}
}
}
